159 lines
5.0 KiB
Python
159 lines
5.0 KiB
Python
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from app.services import project_provisioning
|
|
|
|
|
|
class _FakeGeoServer:
|
|
calls: list[tuple] = []
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *_args):
|
|
return None
|
|
|
|
def check_ready(self):
|
|
self.calls.append(("geoserver_ready",))
|
|
|
|
def workspace_exists(self, workspace):
|
|
self.calls.append(("workspace_exists", workspace))
|
|
return False
|
|
|
|
def create_project_workspace(self, **kwargs):
|
|
self.calls.append(("create_workspace", kwargs))
|
|
return ("junctions", "pipes")
|
|
|
|
def delete_workspace(self, workspace):
|
|
self.calls.append(("delete_workspace", workspace))
|
|
|
|
|
|
def _arrange_preflight(monkeypatch, calls):
|
|
schema_template = project_provisioning.settings.WNDB_SCHEMA_TEMPLATE_DB_NAME
|
|
monkeypatch.setattr(
|
|
project_provisioning,
|
|
"have_project",
|
|
lambda name: name == schema_template,
|
|
)
|
|
monkeypatch.setattr(
|
|
project_provisioning,
|
|
"require_timescale_schema_template",
|
|
lambda: calls.append(("require_timescale_template",)),
|
|
)
|
|
monkeypatch.setattr(
|
|
project_provisioning,
|
|
"ensure_replication_worker_capacity",
|
|
lambda: calls.append(("replication_capacity",)),
|
|
)
|
|
monkeypatch.setattr(
|
|
project_provisioning,
|
|
"timescale_database_exists",
|
|
lambda _name: False,
|
|
)
|
|
monkeypatch.setattr(project_provisioning, "GeoServerAdminClient", _FakeGeoServer)
|
|
|
|
|
|
def test_provision_project_creates_resources_in_dependency_order(monkeypatch):
|
|
calls: list[tuple] = []
|
|
_FakeGeoServer.calls = calls
|
|
_arrange_preflight(monkeypatch, calls)
|
|
monkeypatch.setattr(
|
|
project_provisioning,
|
|
"create_project",
|
|
lambda name: calls.append(("create_business", name)),
|
|
)
|
|
monkeypatch.setattr(
|
|
project_provisioning,
|
|
"network_update",
|
|
lambda path, name: calls.append(("import_model", Path(path).name, name)),
|
|
)
|
|
monkeypatch.setattr(
|
|
project_provisioning,
|
|
"get_project_map_bbox",
|
|
lambda name: (1.0, 2.0, 3.0, 4.0),
|
|
)
|
|
monkeypatch.setattr(
|
|
project_provisioning,
|
|
"create_project_model_template",
|
|
lambda name: calls.append(("create_model_template", name)) or "demo_template",
|
|
)
|
|
monkeypatch.setattr(
|
|
project_provisioning,
|
|
"create_timescale_database",
|
|
lambda name: calls.append(("create_timescale", name)),
|
|
)
|
|
|
|
result = project_provisioning.provision_project_infrastructure(
|
|
code="demo",
|
|
workspace="demo",
|
|
inp_path="model.inp",
|
|
)
|
|
|
|
ordered_actions = [call[0] for call in calls]
|
|
assert ordered_actions.index("create_business") < ordered_actions.index("import_model")
|
|
assert ordered_actions.index("import_model") < ordered_actions.index("create_model_template")
|
|
assert ordered_actions.index("create_model_template") < ordered_actions.index("create_timescale")
|
|
assert ordered_actions.index("create_timescale") < ordered_actions.index("create_workspace")
|
|
assert result.model_template == "demo_template"
|
|
assert result.map_bbox == (1.0, 2.0, 3.0, 4.0)
|
|
|
|
|
|
def test_provision_project_rolls_back_created_resources_in_reverse_order(monkeypatch):
|
|
calls: list[tuple] = []
|
|
_FakeGeoServer.calls = calls
|
|
_arrange_preflight(monkeypatch, calls)
|
|
monkeypatch.setattr(
|
|
project_provisioning,
|
|
"create_project",
|
|
lambda name: calls.append(("create_business", name)),
|
|
)
|
|
monkeypatch.setattr(project_provisioning, "network_update", lambda *_args: None)
|
|
monkeypatch.setattr(
|
|
project_provisioning,
|
|
"get_project_map_bbox",
|
|
lambda _name: (1.0, 2.0, 3.0, 4.0),
|
|
)
|
|
monkeypatch.setattr(
|
|
project_provisioning,
|
|
"create_project_model_template",
|
|
lambda name: calls.append(("create_model_template", name)) or "demo_template",
|
|
)
|
|
monkeypatch.setattr(
|
|
project_provisioning,
|
|
"create_timescale_database",
|
|
lambda _name: (_ for _ in ()).throw(RuntimeError("timescale unavailable")),
|
|
)
|
|
monkeypatch.setattr(
|
|
project_provisioning,
|
|
"delete_project_model_template",
|
|
lambda name: calls.append(("delete_model_template", name)),
|
|
)
|
|
monkeypatch.setattr(
|
|
project_provisioning,
|
|
"delete_project",
|
|
lambda name: calls.append(("delete_business", name)),
|
|
)
|
|
|
|
with pytest.raises(project_provisioning.ProjectProvisioningError) as exc_info:
|
|
project_provisioning.provision_project_infrastructure(
|
|
code="demo",
|
|
workspace="demo",
|
|
inp_path="model.inp",
|
|
)
|
|
|
|
assert exc_info.value.stage == "timescaledb"
|
|
assert [call[0] for call in calls[-2:]] == [
|
|
"delete_model_template",
|
|
"delete_business",
|
|
]
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"code",
|
|
["Demo", "1demo", "demo-project", "demo_template", "postgres"],
|
|
)
|
|
def test_validate_project_code_rejects_unsafe_or_reserved_names(code):
|
|
with pytest.raises(ValueError):
|
|
project_provisioning.validate_project_code(code)
|