0001-hobo-handle-domain-change-59762.patch

Emmanuel Cazenave, 22 décembre 2021 17:46

Voir les différences: en ligne côte à côte

Subject: [PATCH] hobo: handle domain change (#59762)

 tests/test_hobo_sql.py | 294 +++++++++++++++++++++++++++++++++++++++++
 wcs/ctl/check_hobos.py |  68 +++++++---
 2 files changed, 343 insertions(+), 19 deletions(-)
 create mode 100644 tests/test_hobo_sql.py

     import collections
     import copy
     import json
     import os
     import random
     import shutil
     import tempfile
     import zipfile
     import psycopg2
     import pytest
     from quixote import cleanup
     from wcs.ctl.check_hobos import CmdCheckHobos
     from wcs.publisher import WcsPublisher
     from wcs.sql import cleanup_connection
     from .utilities import clean_temporary_pub, create_temporary_pub
     CONFIG = {
         "postgresql": {
             "createdb-connection-params": {"database": "postgres", "user": os.environ['USER']},
             "database-template-name": "%s",
             "user": os.environ['USER'],
+        }
+    }
     WCS_BASE_TENANT = 'wcsteststenant%d' % random.randint(0, 100000)
     WCS_TENANT = '%s.net' % WCS_BASE_TENANT
     WCS_DB_NAME = '%s_net' % WCS_BASE_TENANT
     NEW_WCS_BASE_TENANT = 'wcsteststenant%d' % random.randint(0, 100000)
     NEW_WCS_TENANT = '%s.net' % NEW_WCS_BASE_TENANT
     NEW_WCS_DB_NAME = '%s_net' % NEW_WCS_BASE_TENANT
     HOBO_JSON = {
         'services': [
+            {
                 'title': 'Hobo',
                 'slug': 'hobo',
                 'service-id': 'hobo',
                 'base_url': 'http://hobo.example.net/',
                 'saml-sp-metadata-url': 'http://hobo.example.net/accounts/mellon/metadata/',
             },
+            {
                 'service-id': 'authentic',
                 'saml-idp-metadata-url': 'http://authentic.example.net/idp/saml2/metadata',
                 'template_name': '',
                 'variables': {},
                 'title': 'Authentic',
                 'base_url': 'http://authentic.example.net/',
                 'id': 3,
                 'slug': 'authentic',
                 'secret_key': '12345',
             },
+            {
                 'service-id': 'wcs',
                 'template_name': 'publik.zip',
                 'variables': {'xxx': 'HELLO WORLD'},
                 'title': 'Test wcs',
                 'saml-sp-metadata-url': 'http://%s/saml/metadata' % WCS_TENANT,
                 'base_url': 'http://%s/' % WCS_TENANT,
                 'backoffice-menu-url': 'http://%s/backoffice/menu.json' % WCS_TENANT,
                 'id': 1,
                 'secret_key': 'eiue7aa10nt6e9*#jg2bsfvdgl)cr%4(tafibfjx9i$pgnfj#v',
                 'slug': 'test-wcs',
             },
+            {
                 'service-id': 'combo',
                 'template_name': 'portal-agent',
                 'title': 'Portal Agents',
                 'base_url': 'http://agents.example.net/',
                 'secret_key': 'aaa',
             },
+            {
                 'service-id': 'combo',
                 'template_name': 'portal-user',
                 'title': 'Portal',
                 'base_url': 'http://portal.example.net/',
                 'secret_key': 'bbb',
             },
         ],
         'timestamp': '1431420355.31',
+    }
     @pytest.fixture
     def setuptest():
         cleanup_connection()
         createdb_cfg = CONFIG['postgresql'].get('createdb-connection-params')
         conn = psycopg2.connect(**createdb_cfg)
         conn.set_isolation_level(psycopg2.extensions.ISOLATION_LEVEL_AUTOCOMMIT)
         cursor = conn.cursor()
         for dbname in (WCS_DB_NAME, NEW_WCS_DB_NAME):
             cursor.execute('DROP DATABASE IF EXISTS %s' % dbname)
         pub = create_temporary_pub()
         pub.cfg['language'] = {'language': 'en'}
         hobo_cmd = CmdCheckHobos()
         hobo_cmd.all_services = HOBO_JSON
         WcsPublisher.APP_DIR = tempfile.mkdtemp()
         skeleton_dir = os.path.join(WcsPublisher.APP_DIR, 'skeletons')
         os.mkdir(skeleton_dir)
         with open(os.path.join(skeleton_dir, 'publik.zip'), 'wb') as f:
             with zipfile.ZipFile(f, 'w') as z:
                 z.writestr('config.json', json.dumps(CONFIG))
                 z.writestr('site-options.cfg', '[options]\npostgresql = true')
         yield pub, hobo_cmd
         clean_temporary_pub()
         shutil.rmtree(WcsPublisher.APP_DIR)
         cleanup_connection()
         for dbname in (WCS_DB_NAME, NEW_WCS_DB_NAME):
             cursor.execute('DROP DATABASE IF EXISTS %s' % dbname)
         conn.close()
     def database_exists(database):
         res = False
         cleanup_connection()
         createdb_cfg = CONFIG['postgresql'].get('createdb-connection-params')
         conn = psycopg2.connect(**createdb_cfg)
         conn.set_isolation_level(psycopg2.extensions.ISOLATION_LEVEL_AUTOCOMMIT)
         cursor = conn.cursor()
         cursor.execute("SELECT 1 AS result FROM pg_database WHERE datname='%s'" % database)
         if cursor.fetchall():
             res = True
         conn.close()
         return res
     def test_deploy(setuptest):
         _, hobo_cmd = setuptest
         assert not os.path.exists(os.path.join(WcsPublisher.APP_DIR, 'tenants', WCS_TENANT))
         assert not database_exists(WCS_DB_NAME)
         assert not os.path.exists(os.path.join(WcsPublisher.APP_DIR, 'tenants', NEW_WCS_TENANT))
         assert not database_exists(NEW_WCS_DB_NAME)
         cleanup()
         with open(os.path.join(WcsPublisher.APP_DIR, 'hobo.json'), 'w') as fd:
             fd.write(json.dumps(HOBO_JSON))
         hobo_cmd = CmdCheckHobos()
         base_options = {}
         sub_options_class = collections.namedtuple('Options', ['ignore_timestamp', 'redeploy', 'extra'])
         sub_options = sub_options_class(True, False, None)
         hobo_cmd.execute(
             base_options,
             sub_options,
             ['http://%s/' % WCS_TENANT, os.path.join(WcsPublisher.APP_DIR, 'hobo.json')],
+        )
         assert os.path.exists(os.path.join(WcsPublisher.APP_DIR, 'tenants', WCS_TENANT))
         assert database_exists(WCS_DB_NAME)
         # deploy a new tenant
         with open(os.path.join(WcsPublisher.APP_DIR, 'hobo.json'), 'w') as fd:
             hobo_json = copy.deepcopy(HOBO_JSON)
             wcs_service = hobo_json['services'][2]
             wcs_service['saml-sp-metadata-url'] = ('http://%s/saml/metadata' % WCS_TENANT,)
             wcs_service['base_url'] = 'http://%s/' % NEW_WCS_TENANT
             wcs_service['backoffice-menu-url'] = 'http://%s/backoffice/menu.json' % NEW_WCS_TENANT
             fd.write(json.dumps(hobo_json))
         hobo_cmd.execute(
             base_options,
             sub_options,
             ['http://%s/' % NEW_WCS_TENANT, os.path.join(WcsPublisher.APP_DIR, 'hobo.json')],
+        )
         assert os.path.exists(os.path.join(WcsPublisher.APP_DIR, 'tenants', WCS_TENANT))
         assert database_exists(WCS_DB_NAME)
         assert os.path.exists(os.path.join(WcsPublisher.APP_DIR, 'tenants', NEW_WCS_TENANT))
         assert database_exists(NEW_WCS_DB_NAME)
     def test_deploy_url_change(setuptest):
         _, hobo_cmd = setuptest
         assert not os.path.exists(os.path.join(WcsPublisher.APP_DIR, 'tenants', WCS_TENANT))
         assert not database_exists(WCS_DB_NAME)
         assert not os.path.exists(os.path.join(WcsPublisher.APP_DIR, 'tenants', NEW_WCS_TENANT))
         assert not database_exists(NEW_WCS_DB_NAME)
         cleanup()
         with open(os.path.join(WcsPublisher.APP_DIR, 'hobo.json'), 'w') as fd:
             fd.write(json.dumps(HOBO_JSON))
         hobo_cmd = CmdCheckHobos()
         base_options = {}
         sub_options_class = collections.namedtuple('Options', ['ignore_timestamp', 'redeploy', 'extra'])
         sub_options = sub_options_class(True, False, None)
         hobo_cmd.execute(
             base_options,
             sub_options,
             ['http://%s/' % WCS_TENANT, os.path.join(WcsPublisher.APP_DIR, 'hobo.json')],
+        )
         assert os.path.exists(os.path.join(WcsPublisher.APP_DIR, 'tenants', WCS_TENANT))
         assert database_exists(WCS_DB_NAME)
         # domain change request
         with open(os.path.join(WcsPublisher.APP_DIR, 'hobo.json'), 'w') as fd:
             hobo_json = copy.deepcopy(HOBO_JSON)
             wcs_service = hobo_json['services'][2]
             wcs_service['legacy_urls'] = [
+                {
                     'saml-sp-metadata-url': wcs_service['saml-sp-metadata-url'],
                     'base_url': wcs_service['base_url'],
                     'backoffice-menu-url': wcs_service['backoffice-menu-url'],
+                }
+            ]
             wcs_service['saml-sp-metadata-url'] = ('http://%s/saml/metadata' % WCS_TENANT,)
             wcs_service['base_url'] = 'http://%s/' % NEW_WCS_TENANT
             wcs_service['backoffice-menu-url'] = 'http://%s/backoffice/menu.json' % NEW_WCS_TENANT
             fd.write(json.dumps(hobo_json))
         hobo_cmd.execute(
             base_options,
             sub_options,
             ['http://%s/' % NEW_WCS_TENANT, os.path.join(WcsPublisher.APP_DIR, 'hobo.json')],
+        )
         assert not os.path.exists(os.path.join(WcsPublisher.APP_DIR, 'tenants', WCS_TENANT))
         assert database_exists(WCS_DB_NAME)
         assert os.path.exists(os.path.join(WcsPublisher.APP_DIR, 'tenants', NEW_WCS_TENANT))
         assert not database_exists(NEW_WCS_DB_NAME)
         publisher = WcsPublisher.create_publisher()
         publisher.set_tenant_by_hostname(NEW_WCS_TENANT)
         # check that WCS_DB_NAME is used by NEW_WCS_TENANT
         assert publisher.cfg['postgresql']['database'] == WCS_DB_NAME
         # check that sp configuration is updated
         assert publisher.cfg['sp']['saml2_providerid'] == 'http://%s/saml/metadata' % NEW_WCS_TENANT
     def test_deploy_url_change_old_tenant_dir(setuptest):
         _, hobo_cmd = setuptest
         assert not os.path.exists(os.path.join(WcsPublisher.APP_DIR, 'tenants', WCS_TENANT))
         assert not database_exists(WCS_DB_NAME)
         assert not os.path.exists(os.path.join(WcsPublisher.APP_DIR, 'tenants', NEW_WCS_TENANT))
         assert not database_exists(NEW_WCS_DB_NAME)
         cleanup()
         with open(os.path.join(WcsPublisher.APP_DIR, 'hobo.json'), 'w') as fd:
             fd.write(json.dumps(HOBO_JSON))
         hobo_cmd = CmdCheckHobos()
         base_options = {}
         sub_options_class = collections.namedtuple('Options', ['ignore_timestamp', 'redeploy', 'extra'])
         sub_options = sub_options_class(True, False, None)
         hobo_cmd.execute(
             base_options,
             sub_options,
             ['http://%s/' % WCS_TENANT, os.path.join(WcsPublisher.APP_DIR, 'hobo.json')],
+        )
         assert os.path.exists(os.path.join(WcsPublisher.APP_DIR, 'tenants', WCS_TENANT))
         assert database_exists(WCS_DB_NAME)
         # move tenant to APP_DIR (legacy way)
         os.replace(
             os.path.join(WcsPublisher.APP_DIR, 'tenants', WCS_TENANT),
             os.path.join(WcsPublisher.APP_DIR, WCS_TENANT),
+        )
         assert not os.path.exists(os.path.join(WcsPublisher.APP_DIR, 'tenants', WCS_TENANT))
         # domain change request
         with open(os.path.join(WcsPublisher.APP_DIR, 'hobo.json'), 'w') as fd:
             hobo_json = copy.deepcopy(HOBO_JSON)
             wcs_service = hobo_json['services'][2]
             wcs_service['legacy_urls'] = [
+                {
                     'saml-sp-metadata-url': wcs_service['saml-sp-metadata-url'],
                     'base_url': wcs_service['base_url'],
                     'backoffice-menu-url': wcs_service['backoffice-menu-url'],
+                }
+            ]
             wcs_service['saml-sp-metadata-url'] = ('http://%s/saml/metadata' % WCS_TENANT,)
             wcs_service['base_url'] = 'http://%s/' % NEW_WCS_TENANT
             wcs_service['backoffice-menu-url'] = 'http://%s/backoffice/menu.json' % NEW_WCS_TENANT
             fd.write(json.dumps(hobo_json))
         hobo_cmd.execute(
             base_options,
             sub_options,
             ['http://%s/' % NEW_WCS_TENANT, os.path.join(WcsPublisher.APP_DIR, 'hobo.json')],
+        )
         assert not os.path.exists(os.path.join(WcsPublisher.APP_DIR, WCS_TENANT))
         assert database_exists(WCS_DB_NAME)
         assert os.path.exists(os.path.join(WcsPublisher.APP_DIR, 'tenants', NEW_WCS_TENANT))
         assert not database_exists(NEW_WCS_DB_NAME)
         publisher = WcsPublisher.create_publisher()
         publisher.set_tenant_by_hostname(NEW_WCS_TENANT)
         # check that WCS_DB_NAME is used by NEW_WCS_TENANT
         assert publisher.cfg['postgresql']['database'] == WCS_DB_NAME
         # check that sp configuration is updated
         assert publisher.cfg['sp']['saml2_providerid'] == 'http://%s/saml/metadata' % NEW_WCS_TENANT

             if base_url.endswith('/'):  # wcs doesn't expect a trailing slash
                 service['base_url'] = base_url[:-1]
             new_site = False
             force_spconfig = False
             try:
                 pub.set_tenant_by_hostname(self.get_instance_path(service), skip_sql=True)
                 pub.set_tenant_by_hostname(self.get_instance_path(service.get('base_url')), skip_sql=True)
             except UnknownTenantError:
                 if not os.path.exists(global_tenants_dir):
                     os.mkdir(global_tenants_dir)
                 tenant_app_dir = os.path.join(global_tenants_dir, self.get_instance_path(service))
                 if not os.path.exists(tenant_app_dir):
                 tenant_app_dir = os.path.join(global_tenants_dir, self.get_instance_path(service.get('base_url')))
                 # check in legacy_urls for domain change
                 for legacy_urls in service.get('legacy_urls', []):
                     legacy_base_url = legacy_urls.get('base_url')
                     if legacy_base_url.endswith('/'):  # wcs doesn't expect a trailing slash
                         legacy_base_url = legacy_base_url[:-1]
                     legacy_instance_path = self.get_instance_path(legacy_base_url)
                     try:
                         pub.set_tenant_by_hostname(legacy_instance_path, skip_sql=True)
                         # rename tenant directory
                         for base_dir in (global_app_dir, global_tenants_dir):
                             legacy_tenant_dir = os.path.join(base_dir, legacy_instance_path)
                             if os.path.exists(legacy_tenant_dir):
                                 print('rename tenant directory %s to %s' % (legacy_tenant_dir, tenant_app_dir))
                                 os.rename(legacy_tenant_dir, tenant_app_dir)
                                 break
                         else:
                             print('tenant directory not found')
                             return
                         pub.set_tenant_by_hostname(self.get_instance_path(service.get('base_url')))
                         force_spconfig = True
                         break
                     except UnknownTenantError:
                         pass
                 else:
                     # new tenant
                     print('initializing instance in', tenant_app_dir)
                 os.mkdir(tenant_app_dir)
                 pub.set_tenant_by_hostname(self.get_instance_path(service))
                 if service.get('template_name'):
                     skeleton_filepath = os.path.join(global_app_dir, 'skeletons', service.get('template_name'))
                     if os.path.exists(skeleton_filepath):
                         with open(skeleton_filepath, 'rb') as fd:
                             pub.import_zip(fd)
                 new_site = True
                     os.mkdir(tenant_app_dir)
                     pub.set_tenant_by_hostname(self.get_instance_path(service.get('base_url')))
                     if service.get('template_name'):
                         skeleton_filepath = os.path.join(
                             global_app_dir, 'skeletons', service.get('template_name')
+                        )
                         if os.path.exists(skeleton_filepath):
                             with open(skeleton_filepath, 'rb') as fd:
                                 pub.import_zip(fd)
                     new_site = True
             else:
                 print('updating instance in', pub.app_dir)
                 new_site = False
             try:
                 self.configure_site_options(service, pub, ignore_timestamp=sub_options.ignore_timestamp)
-...
             if new_site:
                 self.configure_sql(service, pub)
             self.update_configuration(service, pub)
             self.configure_authentication_methods(service, pub)
             self.configure_authentication_methods(service, pub, force_spconfig)
             self.update_profile(self.all_services.get('profile', {}), pub)
             # Store hobo.json
-...
                         idp['attribute-mapping'][str(attribute_name)] = str(field_id)
             pub.write_cfg()
         def configure_authentication_methods(self, service, pub):
         def configure_authentication_methods(self, service, pub, force_spconfig=False):
             # look for an identity provider
             idps = [x for x in self.all_services.get('services', []) if x.get('service-id') == 'authentic']
             if not pub.cfg.get('identification'):
-...
                 return
             # initialize service provider side
             if not pub.cfg['sp'].get('publickey'):
             if not pub.cfg['sp'].get('publickey') or force_spconfig:
                 from ..qommon.ident.idp import MethodAdminDirectory
                 spconfig = pub.cfg['sp']
-...
                 pub.cfg['saml_identities']['registration-url'] = str('%saccounts/register/' % idp['base_url'])
                 pub.write_cfg()
         def get_instance_path(self, service):
             parsed_url = urllib.parse.urlsplit(service.get('base_url'))
         def get_instance_path(self, base_url):
             parsed_url = urllib.parse.urlsplit(base_url)
             instance_path = parsed_url.netloc
             if parsed_url.path:
                 instance_path += '+%s' % parsed_url.path.replace('/', '+')
-...
             # determine database name using the instance path
             domain_table_name = (
                 self.get_instance_path(service).replace('-', '_').replace('.', '_').replace('+', '_')
                 self.get_instance_path(service.get('base_url'))
                 .replace('-', '_')
                 .replace('.', '_')
                 .replace('+', '_')
+            )
             if pub.cfg['postgresql'].get('database-template-name'):
+    -

Projet

Général

Profil

Produits Entr'ouvert » w.c.s.

0001-hobo-handle-domain-change-59762.patch