Merge pull request #52 from northpowered/51-test-startup-exceptions

Testing config loader and some startuo events

src/cli/config_loader.py

@@ -3,10 +3,11 @@
 import os
 
 
-def set_config(config_filename) -> None:
+def set_config(config_filename: str, remove_logger: bool = True) -> None:
     from context import config_file
-    from loguru import logger
-    logger.remove()  # Logger supression to beauty CLI output
+    if remove_logger:
+        from loguru import logger
+        logger.remove()  # Logger supression to beauty CLI output
     os.environ['X_FA_CONFIG_FILENAME'] = config_filename
     config_file.set(config_filename)
 

src/config.yaml

@@ -37,8 +37,8 @@ Vault:
   #           1) vault_auth_token from config file
   #           2) root_token from json file
   #keys - simple txt file with unsealing key portions in base64, line by line
-  vault_keyfile_type: "json"
-  vault_unseal_keys: "vault-cluster-vault-2022-07-12T08 03 48.497Z.json"
+  #vault_keyfile_type: "json"
+  #vault_unseal_keys: "vault-cluster-vault-2022-07-12T08 03 48.497Z.json"
   #vault_unseal_keys = "vault-cluster-vault-2022-07-06T18 47 09.634Z.json"
 
 Database:

src/configuration/base.py

@@ -1,22 +1,22 @@
 from pydantic import (BaseModel, ValidationError)
 from loguru import logger
-import os
 
 
 class BaseSectionModel(BaseModel):
 
+    class Config:
+        load_failed: bool = False
+
     def __repr__(self) -> str:
         return f"<FastAPIConfigurationSection object at {hex(id(self))}>"
 
     def load(self, section_data: dict, section_name: str):
         try:
             return self.parse_obj(section_data)
-        except KeyError:
-            logger.error(f'Missed {section_name} section in config file')
-            os._exit(0)
         except ValidationError as ex:
             error = ex.errors()[0]
-            # type: ignore
+            self.Config.load_failed = True
             logger.error(
-                f"{section_name} | {error.get('loc')[0]} | {error.get('msg')}")
-            os._exit(0)
+                f"{section_name} | {error.get('loc')[0]} | {error.get('msg')}"
+            )
+            return None

src/configuration/model.py

@@ -9,14 +9,18 @@
     TelemetrySectionConfiguration
 )
 from .base import BaseSectionModel
-from pydantic import BaseSettings
+from pydantic import BaseSettings, ValidationError
 import configparser
 import toml
 from loguru import logger
+import os
 
 
 class Configuration(BaseSettings):
 
+    class Config:
+        load_failed: bool = False
+
     def __repr__(self) -> str:
         return f"<FastAPIConfiguration object at {hex(id(self))}>"
 
@@ -33,13 +37,16 @@ def load(self, filename: str, filetype: str | None = None):
         try:
             file_extention = filename.split('.')[1]
         except IndexError:
+            self.Config.load_failed = True
             logger.critical('Cannot find config file extention')
         else:
             match file_extention:
                 case 'ini': raw_data = Configuration.ini_reader(filename)
                 case 'toml': raw_data = Configuration.toml_reader(filename)
                 case 'yaml': raw_data = Configuration.yaml_reader(filename)
-                case _: logger.critical('Cannot define config file extention')
+                case _:
+                    self.Config.load_failed = True
+                    logger.critical('Cannot define config file extention')
         self.read_from_dict(raw_data)
         logger.info(f'Configuration was successfully loaded from {filename}')
 
@@ -70,10 +77,13 @@ def read_from_dict(self, raw_data: dict):
         for section_name in self.__fields__:
             section_data: dict = raw_data.get(section_name, dict())
             section: BaseSectionModel = self.__getattribute__(section_name)
+            loaded_section: BaseSectionModel = section.load(
+                section_data,
+                section_name
+            )
+            if not loaded_section:
+                os._exit(1)
             self.__setattr__(
                 section_name,
-                section.load(
-                    section_data,
-                    section_name
-                )
+                loaded_section
             )

src/configuration/sections.py

@@ -24,7 +24,7 @@ class MainSectionConfiguration(BaseSectionModel):
     @validator('application_mode')
     def check_appmode(cls, v):
         assert isinstance(v, str)
-        assert v in ['prod', 'dev']
+        assert v in ['prod', 'dev'], f"Unknown app_mode {v}"
         return v
 
     @validator('log_level')
@@ -329,7 +329,7 @@ class SecuritySectionConfiguration(BaseSectionModel):
     jwt_algorithm: str = "HS256"
     jwt_ttl: int = 3600
 
-    jwt_base_secret_storage: str = 'local'
+    jwt_base_secret_storage: str | None = 'local'
     jwt_base_secret_filename: str = 'secret.key'
     jwt_base_secret_vault_storage_name: str = 'kv'
     jwt_base_secret_vault_secret_name: str = 'jwt'

src/run.py

@@ -14,7 +14,7 @@ def run_app(config_file: str, reload: bool):
         reload (bool): watch file changes and reload server (useful for development)
     """
     print(config_file)
-    set_config(config_file)
+    set_config(config_file, remove_logger=False)
     from configuration import config
     uvicorn.run(
         "app:app",

src/tests/02_startup_events_test.py

@@ -28,6 +28,46 @@ def test_load_endpoint_permissions():
     asyncio.run(load_endpoint_permissions(app))
 
 
-def test_load_base_jwt_secret():
+def test_load_base_jwt_secret_from_config():
     from utils.events import load_base_jwt_secret
     asyncio.run(load_base_jwt_secret())
+
+
+def test_load_base_jwt_secret_from_file():
+    from configuration import config
+    from utils.events import load_base_jwt_secret
+    asyncio.run(
+        load_base_jwt_secret(
+            jwt_base_secret=None,
+            jwt_base_secret_storage='local',
+            jwt_base_secret_filename='src/tests/etc/jwt.txt'
+        )
+    )
+    assert config.Security.get_jwt_base_secret() == 'localfilesecret'
+
+
+def test_load_base_jwt_secret_from_vault():
+    from configuration import config
+    from utils.events import load_base_jwt_secret
+    from utils.vault import Vault
+    vault: Vault = Vault(
+        auth=Vault.VaultAuth(
+            auth_method='token',
+            token='test'
+        )
+    )
+    asyncio.run(
+        load_base_jwt_secret(
+            jwt_base_secret=None,
+            jwt_base_secret_vault_secret_name='jwt',
+            jwt_base_secret_vault_storage_name='kv_test',
+            vault=vault
+        )
+    )
+    assert isinstance(config.Security.get_jwt_base_secret(), str)
+    assert len(config.Security.get_jwt_base_secret()) > 10
+
+
+def test_reload_db_creds():
+    from utils.events import reload_db_creds
+    asyncio.run(reload_db_creds())

src/tests/05_config_test.py

@@ -0,0 +1,43 @@
+from .shared import load_config
+from configuration.sections import ServerSectionConfiguration
+
+
+def test_conf_load_toml():
+    c = load_config('src/config.toml')
+    assert c
+    assert not c.Config.load_failed
+    assert "<FastAPIConfiguration object at" in str(c.__repr__)
+
+
+def test_conf_load_yaml():
+    c = load_config('src/config.yaml')
+    assert c
+    assert not c.Config.load_failed
+    assert "<FastAPIConfiguration object at" in str(c.__repr__)
+
+
+def test_conf_load_ini():
+    c = load_config('src/config.ini')
+    assert c
+    assert not c.Config.load_failed
+    assert "<FastAPIConfiguration object at" in str(c.__repr__)
+
+
+def test_conf_non_existing_config():
+    assert not load_config('non_existing.toml')
+
+
+def test_conf_lost_file_extention():
+    assert load_config('non_existing').Config.load_failed, "Cannot catch lost file extention"
+
+
+def test_conf_unknown_file_extention():
+    assert load_config('non_existing.boroda').Config.load_failed, "Cannot catch unknown file extention"
+
+
+def test_conf_validation_error():
+    bad_data: dict = {
+        'bind_port': 'bar'
+    }
+    section = ServerSectionConfiguration()
+    assert not section.load(bad_data, 'Server')

src/tests/configs/bad_field.toml

@@ -0,0 +1,73 @@
+[Main]
+application_mode = "dev1"
+log_level = "debug"
+log_destination = "stdout"
+log_in_json = 0
+log_sql = 0
+timezone = +3
+enable_swagger = 1
+swagger_doc_url = "/doc"
+swagger_redoc_url = "/redoc"
+enable_security = 1
+
+[AdminGUI]
+admin_enable = 1
+admin_url = "/admin/"
+
+
+[Server]
+bind_address = "localhost"
+bind_port = 8000
+base_url = "example.com"
+
+
+[Vault]
+vault_enable = 1
+vault_host = "localhost"
+vault_port = 8200
+vault_disable_tls = 1
+vault_auth_method = "token"
+vault_token = "test"
+#vault_credentials = 
+vault_try_to_unseal = 1
+#vault_key_type - json | keys
+#json - legacy json file from Vault, created at initialization of Vault instance/cluster
+#       also can contain root_token string, which will be used to access Vault with TOKEN auth_method
+#       Priority:
+#           1) vault_auth_token from config file
+#           2) root_token from json file
+#keys - simple txt file with unsealing key portions in base64, line by line
+#vault_keyfile_type = 
+#vault_unseal_keys = 
+
+[Database]
+db_driver = "postgresql"
+db_host = "127.0.0.1"
+db_port = 5432
+db_name = "test"
+db_username = "test"
+db_password = "test"
+
+db_vault_enable = 1
+db_vault_role = "testrole"
+db_vault_static = 1
+db_vault_storage = "database"
+
+[Telemetry]
+enable = 1
+agent_type = "jaeger"
+agent_host = "localhost"
+agent_port = 6831
+trace_id_length = 12
+
+[Security]
+enable_rbac = 1
+login_with_username = 1
+login_with_email = 1
+jwt_algorithm = "HS256"
+jwt_ttl = 3600
+jwt_base_secret = "dev-secret-from-configfile"
+jwt_base_secret_storage = 'vault'
+jwt_base_secret_filename = 'secret.key1'
+jwt_base_secret_vault_storage_name = 'kv_test'
+jwt_base_secret_vault_secret_name = 'jwt'

src/tests/etc/jwt.txt

@@ -0,0 +1 @@
+localfilesecret

src/tests/payload_models.py

@@ -15,7 +15,7 @@ def create(cls, good_emails: bool = True):
         person: Person = Person('en')
         domains: list[str] = good_email_domains
         if not good_emails:
-            domains = bad_email_domains
+            domains = bad_email_domains  # pragma: no cover
         return UserModel(
             username=person.username(),
             password=person.password(),

src/tests/shared.py

@@ -14,10 +14,10 @@ def clear_migrations_files():
             current_migrations: list = glob.glob(r'*.py', root_dir=directory)
             for current_migration in current_migrations:
                 if current_migration != '__init__.py':
-                    migrations.append(f"{directory}/{current_migration}")
+                    migrations.append(f"{directory}/{current_migration}")  # pragma: no cover
     for migration in migrations:
-        if os.path.isfile(migration):
-            os.remove(migration)
+        if os.path.isfile(migration):  # pragma: no cover
+            os.remove(migration)  # pragma: no cover
 
 
 def prepare_db_with_users(superuser, user):
@@ -31,3 +31,14 @@ def prepare_db_with_users(superuser, user):
     runner.invoke(app, ["aaa", "create", "superuser"],
                   input=superuser.to_cli_input())
     runner.invoke(app, ["aaa", "create", "user"], input=user.to_cli_input())
+
+
+def load_config(filename: str):
+    from configuration.model import Configuration
+    try:
+        config = Configuration()
+        config.load(filename)
+    except FileNotFoundError:
+        return None
+    else:
+        return config