diff --git a/backend/tests/helpers/constants.py b/backend/tests/helpers/constants.py index 5f61e866db..bd9ec136ee 100644 --- a/backend/tests/helpers/constants.py +++ b/backend/tests/helpers/constants.py @@ -2,7 +2,7 @@ from infrahub_sdk.utils import str_to_bool -INFRAHUB_USE_TEST_CONTAINERS = str_to_bool(os.getenv("INFRAHUB_USE_TEST_CONTAINERS", "true")) +INFRAHUB_USE_TEST_CONTAINERS = str_to_bool(os.getenv("INFRAHUB_USE_TEST_CONTAINERS", "false")) PORT_NATS = 4222 PORT_REDIS = 6379 PORT_CLIENT_RABBITMQ = 5672 diff --git a/backend/tests/helpers/schema/__init__.py b/backend/tests/helpers/schema/__init__.py index 4b5610843b..f4fc7d9bcb 100644 --- a/backend/tests/helpers/schema/__init__.py +++ b/backend/tests/helpers/schema/__init__.py @@ -27,15 +27,13 @@ async def load_schema( db: InfrahubDatabase, schema: SchemaRoot, branch_name: str | None = None, update_db: bool = False ) -> None: - default_branch_name = registry.default_branch - branch_schema = registry.schema.get_schema_branch(name=branch_name or default_branch_name) - tmp_schema = branch_schema.duplicate() - tmp_schema.load_schema(schema=schema) - tmp_schema.process() - + branch_name = branch_name or registry.default_branch + branch_schema = registry.schema.get_schema_branch(name=branch_name) + await registry.schema.register_schema(schema=schema, branch=branch_name) await registry.schema.update_schema_branch( - schema=tmp_schema, db=db, branch=branch_name or default_branch_name, update_db=update_db + schema=branch_schema.duplicate(), db=db, branch=branch_name, update_db=update_db ) + registry.get_branch_from_registry(branch_name).update_schema_hash() __all__ = [ diff --git a/backend/tests/integration/shared.py b/backend/tests/integration/shared.py index 9d64d10c76..d8abbda2c8 100644 --- a/backend/tests/integration/shared.py +++ b/backend/tests/integration/shared.py @@ -1,15 +1,9 @@ from typing import Any, Dict -from infrahub.core import registry from infrahub.core.schema import SchemaRoot from infrahub.database import InfrahubDatabase +from tests.helpers.schema import load_schema -async def load_schema(db: InfrahubDatabase, schema: Dict[str, Any]): - default_branch_name = registry.default_branch - branch_schema = registry.schema.get_schema_branch(name=default_branch_name) - tmp_schema = branch_schema.duplicate() - tmp_schema.load_schema(schema=SchemaRoot(**schema)) - tmp_schema.process() - - await registry.schema.update_schema_branch(schema=tmp_schema, db=db, branch=default_branch_name, update_db=True) +async def load_schema_(db: InfrahubDatabase, schema: Dict[str, Any]): + await load_schema(db=db, schema=SchemaRoot(**schema), update_db=True)