from __future__ import annotations import ast import subprocess import sys import tempfile import unittest from pathlib import Path ROOT = Path(__file__).resolve().parents[1] class InfrastructureConfigTests(unittest.TestCase): def test_structlog_event_is_not_passed_twice(self) -> None: log_methods = {"debug", "info", "warning", "error", "exception", "critical"} for source_path in (ROOT / "api-backend/app").rglob("*.py"): tree = ast.parse(source_path.read_text(encoding="utf-8"), filename=str(source_path)) for node in ast.walk(tree): if ( isinstance(node, ast.Call) and isinstance(node.func, ast.Attribute) and node.func.attr in log_methods and node.args and any(keyword.arg == "event" for keyword in node.keywords) ): self.fail(f"duplicate structlog event in {source_path}:{node.lineno}") def test_root_compose_uses_only_infra_fragments(self) -> None: compose = (ROOT / "docker-compose.yml").read_text(encoding="utf-8") for fragment in ( "infra/compose/application.yml", "nginx/docker-compose.yml", "redis/docker-compose.yml", "observability/docker-compose.yml", "deployment/docker-compose.jobs.yml", ): self.assertIn(fragment, compose) self.assertNotIn("postgres:", compose.lower()) def test_external_dependencies_use_dedicated_egress_network(self) -> None: root = (ROOT / "docker-compose.yml").read_text(encoding="utf-8") self.assertIn("egress:\n name: han-chat-egress", root) application = (ROOT / "infra/compose/application.yml").read_text(encoding="utf-8") self.assertIn("networks: [backend, observability, egress]", application) self.assertIn("networks: [public, backend, observability]", application) jobs = (ROOT / "deployment/docker-compose.jobs.yml").read_text(encoding="utf-8") self.assertEqual(jobs.count("networks: [backend, egress]"), 4) observability = (ROOT / "observability/docker-compose.yml").read_text(encoding="utf-8") self.assertIn("networks: [observability, backend, egress]", observability) redis = (ROOT / "redis/docker-compose.yml").read_text(encoding="utf-8") self.assertNotIn("egress", redis) def test_only_nginx_fragment_publishes_ports(self) -> None: forbidden = ( ROOT / "infra/compose/application.yml", ROOT / "redis/docker-compose.yml", ROOT / "observability/docker-compose.yml", ROOT / "deployment/docker-compose.jobs.yml", ) for path in forbidden: self.assertNotIn("\n ports:", path.read_text(encoding="utf-8"), path) nginx = (ROOT / "nginx/docker-compose.yml").read_text(encoding="utf-8") self.assertEqual(nginx.count("\n ports:"), 1) self.assertIn('NGINX_HTTP_PORT:-80}:80', nginx) self.assertIn('NGINX_HTTPS_PORT:-443}:443', nginx) def test_nginx_internal_denies_precede_spa(self) -> None: site = (ROOT / "nginx/templates/site-tls.conf.template").read_text(encoding="utf-8") compose = (ROOT / "nginx/docker-compose.yml").read_text(encoding="utf-8") config = (ROOT / "nginx/nginx.conf.template").read_text(encoding="utf-8") proxy_common = (ROOT / "nginx/snippets/proxy-common.conf").read_text(encoding="utf-8") proxy_keycloak = (ROOT / "nginx/snippets/proxy-keycloak.conf").read_text( encoding="utf-8" ) internal = site.index("location ^~ /internal/") api = site.index("location ^~ /api/") frontend = site.index("include /etc/nginx/generated/frontend-location.conf") self.assertLess(internal, api) self.assertLess(api, frontend) self.assertIn("location = /api/v1/realtime", site) self.assertNotIn("message-safety:", site) self.assertIn("uid=0,gid=0", compose) self.assertIn('cap_add: ["CHOWN", "NET_BIND_SERVICE", "SETUID", "SETGID"]', compose) self.assertTrue(config.startswith("user nginx;\n")) self.assertIn("proxy_read_timeout 30s;", config) self.assertNotIn("proxy_read_timeout", proxy_common) self.assertNotIn("proxy_send_timeout", proxy_common) self.assertIn("include /etc/nginx/snippets/proxy-keycloak.conf;", site) self.assertIn("location = /auth/callback", site) self.assertIn("location ^~ /auth/resources/", site) self.assertIn("location ^~ /auth/realms/", site) self.assertNotIn("security-headers.conf", proxy_keycloak) self.assertNotIn("X-Frame-Options", proxy_keycloak) def test_redis_persistence_acl_and_no_host_port(self) -> None: config = (ROOT / "redis/redis.conf").read_text(encoding="utf-8") acl = (ROOT / "redis/users.acl.template").read_text(encoding="utf-8") self.assertIn("appendonly yes", config) self.assertIn("appendfsync everysec", config) self.assertIn("save 900 1", config) self.assertIn("user default off", acl) self.assertIn("~han:api:*", acl) self.assertIn("~han:safety:*", acl) def test_otel_has_redaction_and_persistent_queue(self) -> None: config = (ROOT / "observability/otel-collector.yaml").read_text(encoding="utf-8") for forbidden_attribute in ( "http.request.header.authorization", "http.request.header.cookie", "url.query", "db.statement", "messaging.message.body", ): self.assertIn(forbidden_attribute, config) self.assertIn("storage: file_storage", config) self.assertIn("retry_on_failure:", config) def test_settings_cli_and_workers_are_deployable(self) -> None: cli = ROOT / "api-backend/app/cli" self.assertTrue((cli / "__init__.py").is_file()) self.assertTrue((cli / "seed_settings.py").is_file()) self.assertTrue((cli / "validate_settings.py").is_file()) jobs = (ROOT / "deployment/docker-compose.jobs.yml").read_text(encoding="utf-8") self.assertIn("python -m app.cli.seed_settings", jobs) self.assertIn("python -m app.cli.validate_settings", jobs) application = (ROOT / "infra/compose/application.yml").read_text(encoding="utf-8") self.assertIn("\n frontend-static:", application) self.assertIn("frontend-test-site", application) self.assertIn("frontend-static:/output", application) for service, command in ( ("delivery-worker:", "han-delivery-worker"), ("safety-recovery-worker:", "han-safety-worker"), ("cleanup-worker:", "han-cleanup-worker"), ): self.assertIn(f"\n {service}", application) self.assertIn(f'command: ["{command}"]', application) self.assertNotIn("\n ports:", application) nginx = (ROOT / "nginx/docker-compose.yml").read_text(encoding="utf-8") self.assertIn("frontend-static: {condition: service_completed_successfully}", nginx) def test_smoke_script_does_not_source_env_as_shell(self) -> None: smoke = (ROOT / "deployment/scripts/smoke.sh").read_text(encoding="utf-8") self.assertNotIn('. "./$ENV_FILE"', smoke) for variable in ("PUBLIC_HOST", "PUBLIC_WEB_URL", "KEYCLOAK_REALM"): self.assertIn(f"{variable}=$(env_value {variable})", smoke) def test_expo_public_environment_uses_static_property_access(self) -> None: config = (ROOT / "frontend-test-site/src/config.ts").read_text(encoding="utf-8") auth = (ROOT / "frontend-test-site/src/auth.ts").read_text(encoding="utf-8") callback = (ROOT / "frontend-test-site/app/auth/callback.tsx").read_text( encoding="utf-8" ) self.assertNotIn("process.env[name]", config) self.assertIn("process.env.EXPO_PUBLIC_API_BASE_URL", config) self.assertIn("process.env.EXPO_PUBLIC_AUTH_BASE_URL", config) self.assertIn('path: "auth/callback"', auth) self.assertIn('window.location.assign(url)', auth) self.assertIn("completionStarted.current", callback) self.assertTrue((ROOT / "frontend-test-site/app/auth/callback.tsx").is_file()) def test_alembic_escapes_percent_encoded_dsn_options(self) -> None: for relative_path in ( "api-backend/alembic/env.py", "bitrix-local-app/alembic/env.py", "bitrix-sync/alembic/env.py", ): env_script = (ROOT / relative_path).read_text(encoding="utf-8") self.assertIn('.replace("%", "%%")', env_script, relative_path) self.assertIn("create_postgres_engine", env_script, relative_path) def test_contact_sync_qualifies_pgcrypto_digest(self) -> None: initial = ( ROOT / "api-backend/alembic/versions/0001_initial_han_app.py" ).read_text(encoding="utf-8") fix = ( ROOT / "api-backend/alembic/versions/0002_qualify_pgcrypto_digest.py" ).read_text(encoding="utf-8") main = (ROOT / "api-backend/app/main.py").read_text(encoding="utf-8") self.assertIn("public.digest(", initial) self.assertIn("public.digest(", fix) self.assertIn('down_revision: str | None = "0001_initial"', fix) self.assertIn('revision != "0003_consent_audit"', main) def test_consent_audit_migration_supports_existing_and_fresh_databases(self) -> None: migration = ( ROOT / "api-backend/alembic/versions/0003_consent_device_audit_context.py" ).read_text(encoding="utf-8") self.assertIn("ADD COLUMN IF NOT EXISTS device_json", migration) self.assertIn("DROP COLUMN IF EXISTS ip", migration) self.assertIn('down_revision: str | None = "0002_pgcrypto_digest"', migration) def test_keycloak_management_health_and_bridge_environment(self) -> None: standalone = (ROOT / "keycloak/docker-compose.yml").read_text(encoding="utf-8") self.assertIn("GET /auth/health/ready", standalone) dockerfile = (ROOT / "keycloak/Dockerfile").read_text(encoding="utf-8") self.assertLess( dockerfile.index("COPY realm ./realm"), dockerfile.index("mvn -B -ntp clean verify"), ) application = (ROOT / "infra/compose/application.yml").read_text(encoding="utf-8") self.assertIn("GET /auth/health/ready", application) for variable in ( "KC_DB_SCHEMA", "KEYCLOAK_OTP_MOCK_ENABLED", "KEYCLOAK_OTP_MOCK_CODE", "KEYCLOAK_OTP_HMAC_KEY", "KEYCLOAK_OTP_TTL_SEC", "KEYCLOAK_OTP_MAX_VERIFY_ATTEMPTS", "KEYCLOAK_OTP_SETTINGS_MAX_STALE_SEC", "KEYCLOAK_SETTINGS_BRIDGE_URL", "KEYCLOAK_SETTINGS_BRIDGE_TOKEN", ): self.assertIn(f" {variable}:", application) def test_env_validator_accepts_materialized_example(self) -> None: example = (ROOT / ".env.example").read_text(encoding="utf-8") self.assertNotIn("options=-csearch_path", example) self.assertNotIn("currentSchema=", example) self.assertIn("KEYCLOAK_DB_SCHEMA=keycloak", example) self.assertIn("HAN_PG_PORT=5433", example) for required in ( "CURSOR_HMAC_SECRET=", "BITRIX_TOKEN_ENCRYPTION_KEY=", "KEYCLOAK_INTERNAL_URL=http://keycloak:8080/auth", ): self.assertIn(required, example) materialized = example.replace("change-me", "0123456789abcdef0123456789abcdef") materialized = materialized.replace( "KEYCLOAK_OTP_MOCK_RISK_ACCEPTED=false", "KEYCLOAK_OTP_MOCK_RISK_ACCEPTED=true", ) with tempfile.TemporaryDirectory() as directory: env_file = Path(directory) / ".env" env_file.write_text(materialized, encoding="utf-8") result = subprocess.run( [sys.executable, str(ROOT / "scripts/validate-env"), str(env_file)], text=True, capture_output=True, check=False, ) self.assertEqual(result.returncode, 0, result.stderr) if __name__ == "__main__": unittest.main()