"""E2E tests for HTTP statuses reachable against a locally-running server with no external backends. This service is a "200-envelope" API: most data endpoints return HTTP 200 and carry a logical result code in the JSON body. Only reqparse endpoints and the App Runner blueprint return real HTTP 4xx/5xx. These assertions reflect the actual HTTP contract (see docs/openapi.yaml).""" import pytest pytestmark = pytest.mark.e2e # ── Core / health ──────────────────────────────────────────────────────────── def test_root(api): api.check("GET", "/", "/", 200) def test_ping(api): api.check("GET", "/test", "/test", 200) def test_l7check(api): api.check("GET", "/monitor/l7check", "/monitor/l7check", 200) def test_headers(api): api.check("GET", "/headers", "/headers", 200) def test_swagger_json(api): api.check("GET", "/swagger.json", "/swagger.json", 200) # ── reqparse endpoints (JSON body required → 415 otherwise) ────────────────── def test_run_cmd_ok(api): api.check("POST", "/run_cmd", "/run_cmd", 200, json={"cmd": "echo hi"}) def test_run_cmd_empty(api): api.check("POST", "/run_cmd", "/run_cmd", 400, json={"cmd": ""}) def test_run_cmd_wrong_content_type(api): api.check("POST", "/run_cmd", "/run_cmd", 415, data={"cmd": "echo hi"}) def test_webhook(api): api.check("POST", "/webhook", "/webhook", 200, json={"hello": "world"}) def test_run_iperf_bad_protocol(api): api.check( "POST", "/run_iperf", "/run_iperf", 400, json={"host_ip": "1.1.1.1", "host_port": "5201", "duration": 1, "protocol": "icmp"}, ) def test_run_iperf_wrong_content_type(api): api.check("POST", "/run_iperf", "/run_iperf", 415, data={"protocol": "tcp"}) def test_athenz_missing_args(api): api.check("POST", "/athenz_access_token", "/athenz_access_token", 400, json={}) def test_athenz_wrong_content_type(api): api.check( "POST", "/athenz_access_token", "/athenz_access_token", 415, data={"account_name": "a"} ) def test_athenz_internal_error(api): api.check( "POST", "/athenz_access_token", "/athenz_access_token", 500, json={"account_name": "acct", "provider_domain": "p.stage.proj"}, ) def test_test_egp_missing_env(api): api.check( "POST", "/test_egp", "/test_egp", 500, json={"egp_endpoint": "", "egp_token": "", "remote_endpoint": ""}, ) # ── markdown catch-all ─────────────────────────────────────────────────────── def test_markdown_existing(api): api.check("GET", "/{file_path}", "/Home.md", 200) def test_markdown_missing(api): api.check("GET", "/{file_path}", "/no-such-file.md", 404) # ── UI pages ───────────────────────────────────────────────────────────────── def test_fos_ui_template_error(api): api.check("GET", "/test_fos_ui/", "/test_fos_ui/", 500) @pytest.mark.parametrize( "path", [ "/test_fos_crud_ui/", "/test_fos_dual_env_ui/", "/dbs_test_ui/", "/test_mysql_dual_env_ui/", "/function_test_ui/", "/app_runner_test_ui/", "/egp_test_ui/", ], ) def test_ui_pages(api, path): api.check("GET", path, path, 200) # ── FOS (200-envelope; logical code in body) ───────────────────────────────── def test_put_test_fos(api): api.check( "PUT", "/test_fos/buckets/{bucket_name}/{object_name}", "/test_fos/buckets/b/o", 200, json={"data": "x"}, ) def test_get_fos_object(api): api.check("GET", "/get_fos_object/{bucket_name}/{object_name}", "/get_fos_object/b/o", 200) def test_test_fos_crud(api): api.check( "POST", "/test_fos_crud/{bucket_name}/{object_name}", "/test_fos_crud/b/o", 200, json={"env": "prod", "data": "x"}, ) def test_create_fos_object_missing_data(api): api.check( "PUT", "/create_fos_object/{bucket_name}/{object_name}", "/create_fos_object/b/o", 400, json={"env": "prod"}, ) def test_list_objects(api): api.check("GET", "/list_objects/{bucket_name}", "/list_objects/b", 200) def test_update_bucket(api): api.check( "POST", "/update_bucket/{bucket_name}/{action}", "/update_bucket/b/frobnicate", 200, json={"env": "prod"}, ) def test_write_dual_env(api): api.check( "POST", "/write_to_dual_env/{bucket_name}/{object_name}", "/write_to_dual_env/b/o", 200, json={"data": "x"}, ) def test_list_dual_env(api): api.check("GET", "/list_from_dual_env/{bucket_name}", "/list_from_dual_env/b", 200) # ── Databases (200-envelope) ────────────────────────────────────────────────── def test_connect(api): api.check("POST", "/connect", "/connect", 200, json={"product": "bogus"}) def test_execute_query(api): api.check( "POST", "/execute_query", "/execute_query", 200, json={"product": "bogus", "query": "x"} ) def test_execute_opensearch(api): api.check( "POST", "/execute_opensearch_query", "/execute_opensearch_query", 200, json={"service": "s", "user": "u", "password": "p", "method": "GET", "endpoint": "/_cat"}, ) def test_connect_dual_mysql(api): api.check("POST", "/connect_dual_mysql", "/connect_dual_mysql", 200, data="") def test_execute_dual_mysql(api): api.check("POST", "/execute_dual_mysql_query", "/execute_dual_mysql_query", 200, data="") # ── App Runner blueprint (real HTTP codes) ─────────────────────────────────── def test_env_vars_all(api): api.check("GET", "/api/app_runner/env_vars", "/api/app_runner/env_vars", 200) def test_env_var_found(api): api.check("GET", "/api/app_runner/env_vars/{key}", "/api/app_runner/env_vars/PATH", 200) def test_env_var_missing(api): api.check( "GET", "/api/app_runner/env_vars/{key}", "/api/app_runner/env_vars/CQA_DEFINITELY_MISSING", 404, ) def test_get_access_token_no_domain(api): api.check("GET", "/api/app_runner/get_access_token", "/api/app_runner/get_access_token", 400) def test_get_access_token_error(api): api.check( "GET", "/api/app_runner/get_access_token", "/api/app_runner/get_access_token?product_domain=dom", 500, ) def test_send_request_no_host(api): api.check("GET", "/api/app_runner/send_request", "/api/app_runner/send_request", 400) def test_send_request_unreachable_host(api): api.check( "GET", "/api/app_runner/send_request", "/api/app_runner/send_request?host=http://127.0.0.1:1", 500, ) def test_validate_volume_missing_params(api): api.check( "POST", "/api/app_runner/validate_volume", "/api/app_runner/validate_volume", 400, json={} ) def test_validate_volume_file_exists(api): api.check( "POST", "/api/app_runner/validate_volume", "/api/app_runner/validate_volume", 200, json={"key": "run_cqa_test_app.py", "path": "."}, ) def test_validate_volume_file_missing(api): api.check( "POST", "/api/app_runner/validate_volume", "/api/app_runner/validate_volume", 404, json={"key": "nope.txt", "path": "/tmp"}, ) # ── Locally reachable when the server is started via run_cqa_test_app.py ────── def test_explorer(api): # The :10346 explorer subprocess is started by the entrypoint. api.check("GET", "/explorer/{path}", "/explorer/", 200) def test_send_request_ok(api): # Point host at the separate :10346 process (not self) to avoid a # single-threaded-server deadlock. api.check( "GET", "/api/app_runner/send_request", "/api/app_runner/send_request?host=http://localhost:10346/", 200, ) def test_test_egp_result(api): # Any non-empty config reaches HTTP 200 (errors are returned in the body). api.check( "POST", "/test_egp", "/test_egp", 200, json={ "egp_endpoint": "http://localhost:19999", "egp_token": "x", "remote_endpoint": "http://localhost:10346/", }, ) def test_start_least_conn(api): api.check("POST", "/start_least_conn_processes", "/start_least_conn_processes", 200)