108 lines
4.3 KiB
Python
108 lines
4.3 KiB
Python
from pathlib import Path
|
|
|
|
from scripts import load_test, track7_check
|
|
|
|
|
|
def test_load_test_profiles_and_mix():
|
|
profile = load_test.resolve_profile("target_500_500")
|
|
compat_profile = load_test.resolve_profile("baseline_100_100", voice=120, digital=80)
|
|
mix = load_test.build_mix_profile(profile, include_read_traffic=True)
|
|
|
|
assert profile["voice"] == 500
|
|
assert profile["digital"] == 500
|
|
assert compat_profile["voice"] == 120
|
|
assert compat_profile["digital"] == 80
|
|
assert mix["channels"]["voice"] == 500
|
|
assert mix["channels"]["telegram"] == 200
|
|
assert mix["channels"]["webchat"] == 150
|
|
assert mix["channels"]["email"] == 150
|
|
assert "/proxy/reporting/reports/kpi" in mix["read_endpoints"]
|
|
|
|
|
|
def test_load_test_request_builders_use_real_adapter_routes():
|
|
telegram_request = load_test.build_request("telegram", 1)
|
|
webchat_request = load_test.build_request("webchat", 1)
|
|
email_request = load_test.build_request("email", 1)
|
|
|
|
assert telegram_request[1] == "/proxy/telegram/integrations/telegram/webhook"
|
|
assert webchat_request[1] == "/proxy/webchat/integrations/webchat/messages"
|
|
assert email_request[1] == "/proxy/email/integrations/email/messages"
|
|
|
|
|
|
def test_load_test_report_writing_and_threshold_evaluation(tmp_path: Path):
|
|
mix = load_test.build_mix_profile(load_test.resolve_profile("baseline_100_100"), include_read_traffic=False)
|
|
results = [
|
|
{"ts": 1.0, "target": "voice", "status_code": 200, "latency_seconds": 0.5, "ok": True},
|
|
{"ts": 2.0, "target": "telegram", "status_code": 200, "latency_seconds": 0.7, "ok": True},
|
|
{"ts": 3.0, "target": "voice", "status_code": 502, "latency_seconds": 1.2, "ok": False},
|
|
]
|
|
summary = load_test.build_summary(
|
|
results,
|
|
mix,
|
|
{
|
|
"require_success_rate": 50.0,
|
|
"require_p95_seconds": 2.0,
|
|
"require_p99_seconds": 3.0,
|
|
},
|
|
"2026-03-02T10:00:00",
|
|
"2026-03-02T10:01:00",
|
|
)
|
|
|
|
load_test.write_report(tmp_path, summary, results, [{"error": "bad gateway"}])
|
|
|
|
assert summary["passed"] is True
|
|
assert (tmp_path / "summary.json").exists()
|
|
assert (tmp_path / "latency_samples.csv").exists()
|
|
assert (tmp_path / "error_samples.json").exists()
|
|
assert (tmp_path / "mix_profile.json").exists()
|
|
|
|
|
|
def test_track7_check_evaluate_summary_flags_failures():
|
|
summary = {
|
|
"results": {
|
|
"success_rate": 97.5,
|
|
"p95_seconds": 2.5,
|
|
"p99_seconds": 4.1,
|
|
"five_xx_rate": 0.8,
|
|
}
|
|
}
|
|
|
|
issues = track7_check.evaluate_summary(
|
|
summary,
|
|
require_success_rate=99.0,
|
|
require_p95_seconds=2.0,
|
|
require_p99_seconds=3.5,
|
|
)
|
|
|
|
assert any("Success rate" in issue for issue in issues)
|
|
assert any("P95" in issue for issue in issues)
|
|
assert any("P99" in issue for issue in issues)
|
|
assert any("5xx" in issue for issue in issues)
|
|
|
|
|
|
def test_track7_artifacts_present_in_templates_and_runtime():
|
|
root = Path(__file__).resolve().parents[1]
|
|
db_py = (root / "services" / "shared" / "db.py").read_text(encoding="utf-8")
|
|
values_yaml = (root / "deployment" / "helm" / "values.yaml").read_text(encoding="utf-8")
|
|
services_yaml = (root / "deployment" / "helm" / "templates" / "services.yaml").read_text(encoding="utf-8")
|
|
migration_job_yaml = (root / "deployment" / "helm" / "templates" / "migration-job.yaml").read_text(encoding="utf-8")
|
|
|
|
assert "DB_POOL_SIZE" in db_py
|
|
assert "DB_MAX_OVERFLOW" in db_py
|
|
assert "profiles:" in values_yaml
|
|
assert "scale500:" in values_yaml
|
|
assert "databaseUrl:" in values_yaml
|
|
assert "migrations:" in values_yaml
|
|
assert "HorizontalPodAutoscaler" in services_yaml
|
|
assert "PodDisruptionBudget" in services_yaml
|
|
assert "startupProbe" in services_yaml
|
|
assert "livenessProbe" in services_yaml
|
|
assert "DATABASE_URL" in services_yaml
|
|
assert "ai-orchestrator-service:" in values_yaml
|
|
assert "ai-voice-runtime-service:" in values_yaml
|
|
assert "AI_VOICE_RUNTIME_SERVICE_URL" in services_yaml
|
|
assert "AI_VOICE_QUEUE_CONFIG_JSON" in services_yaml
|
|
assert '"helm.sh/hook": pre-install,pre-upgrade' in migration_job_yaml
|
|
assert "scripts/migrate_core_db.py" in migration_job_yaml
|
|
assert "SCHEMA_MANAGEMENT_MODE" in migration_job_yaml
|