prole/tests/installer/test_prod_config.py

121 lines
4.5 KiB
Python

from pathlib import Path
import sys
# Ensure `installer` is importable when tests are invoked directly via `pytest`.
PROJECT_ROOT = Path(__file__).resolve().parents[2]
if str(PROJECT_ROOT) not in sys.path:
sys.path.insert(0, str(PROJECT_ROOT))
from knoe.core.prod_config import (
ProdConfigApi,
canonical_yaml,
default_prod_config,
opentofu_vars,
validate_prod_config,
)
def test_default_prod_config_uses_knoey_knoe_conventions():
cfg = default_prod_config()
assert cfg.kind == "KnoeProductionConfig"
assert cfg.metadata.ecosystemId == 0
assert cfg.metadata.name == "knoey-root"
assert cfg.kubernetes.namespace == "ecosystem-0"
assert cfg.cloud.clusterName == "knoe-prod"
assert cfg.database.clusterName == "knoe-db"
assert cfg.database.pgdataStorageClass == "premium-rwo"
assert cfg.database.walStorageClass == "premium-rwo"
assert cfg.database.appDatabase == "knoey"
assert cfg.database.metaDatabase == "knoe_meta"
assert cfg.backups.backupBucket == "knoe-0-backups"
assert cfg.backups.walBucket == "knoe-0-wal"
assert cfg.routing.frontdoorHost == "knoey.com"
assert cfg.routing.platformDomain == "knoe.dev"
assert cfg.routing.apiHost == "api.knoe.dev"
assert cfg.routing.studioHost == "db.0.knoe.dev"
assert cfg.routing.gitHost == "git.knoe.dev"
assert cfg.migration.sourceEnvironment == "knoe.org"
def test_validation_requires_project_and_enforces_ecosystem_and_positive_numbers():
cfg = default_prod_config()
cfg.cloud.projectId = ""
cfg.metadata.ecosystemId = 1
cfg.database.instances = 0
cfg.database.storageSizeGi = -1
cfg.backups.retentionDays = 0
cfg.migration.sourcePort = 0
cfg.routing.frontdoorHost = "not_a_domain"
cfg.routing.apiHost = "not_a_domain"
errors, _warnings = validate_prod_config(cfg)
assert "metadata.ecosystemId must be fixed to 0" in errors
assert "spec.cloud.projectId is required" in errors
assert "spec.database.instances must be a positive number" in errors
assert "spec.database.storageSizeGi must be a positive number" in errors
assert "spec.backups.retentionDays must be a positive number" in errors
assert "spec.migration.sourcePort must be a positive number" in errors
assert "spec.routing.frontdoorHost must be a valid domain" in errors
assert "spec.routing.apiHost must be a valid domain" in errors
def test_yaml_and_opentofu_preview_contains_expected_values():
cfg = default_prod_config()
cfg.cloud.projectId = "my-gcp-project"
cfg.database.pgdataStorageClass = "standard-rwo"
cfg.database.walStorageClass = "premium-rwo"
cfg.database.pgdataSizeGi = 120
cfg.database.walSizeGi = 32
yaml_preview = canonical_yaml(cfg)
tf_vars = opentofu_vars(cfg)
assert "kind: KnoeProductionConfig" in yaml_preview
assert "ecosystemId: 0" in yaml_preview
assert "projectId: my-gcp-project" in yaml_preview
assert "pgdataStorageClass: standard-rwo" in yaml_preview
assert "walStorageClass: premium-rwo" in yaml_preview
assert "frontdoorHost: knoey.com" in yaml_preview
assert "apiHost: api.knoe.dev" in yaml_preview
assert "studioHost: db.0.knoe.dev" in yaml_preview
assert "gitHost: git.knoe.dev" in yaml_preview
assert tf_vars["project_id"] == "my-gcp-project"
assert tf_vars["namespace"] == "ecosystem-0"
assert tf_vars["cnpg_cluster_name"] == "knoe-db"
assert tf_vars["db_pgdata_storage_class"] == "standard-rwo"
assert tf_vars["db_wal_storage_class"] == "premium-rwo"
assert tf_vars["db_pgdata_size_gi"] == 120
assert tf_vars["db_wal_size_gi"] == 32
assert tf_vars["supabase_api_host"] == "api.knoe.dev"
assert tf_vars["supabase_studio_host"] == "db.0.knoe.dev"
assert tf_vars["gitlab_host"] == "git.knoe.dev"
def test_prod_config_api_plan_apply_status_and_logs_flow():
api = ProdConfigApi()
payload = api.get_prod_config()
payload["spec"]["cloud"]["projectId"] = "my-gcp-project"
put_result = api.put_prod_config(payload)
assert put_result["ok"] is True
assert put_result["errors"] == []
plan_result = api.post_plan(payload)
assert plan_result["ok"] is True
assert "kind: KnoeProductionConfig" in plan_result["yaml"]
assert "project_id" in plan_result["opentofuVars"]
assert plan_result["installPlan"]
apply_result = api.post_apply(payload)
assert apply_result["ok"] is True
assert apply_result["errors"] == []
status = api.get_status()
logs = api.get_logs()
assert status["phase"] == "apply"
assert status["state"] == "applied"
assert logs["logs"]