From 5a99423d32599716720ca75598657e3dd2649c0a Mon Sep 17 00:00:00 2001 From: Manish Kumar Date: Fri, 18 Sep 2026 11:13:26 -0500 Subject: [PATCH] docs: improve test suite documentation Co-Authored-By: Claude Sonnet 5 --- docker-socket-proxy/tests/test_policy.py | 86 +++++++++++++++++++ .../tests/test_proxy_connection_handling.py | 17 ++++ .../tests/test_proxy_live_integration.py | 23 +++++ tests/integration/conftest.py | 11 +++ tests/integration/test_auth_integration.py | 52 +++++++++++ tests/integration/test_auth_registration.py | 15 ++++ tests/integration/test_auth_revocation.py | 11 +++ tests/integration/test_conftest_fixtures.py | 24 ++++++ tests/integration/test_cross_app_sso.py | 12 +++ tests/integration/test_lims_integration.py | 44 ++++++++++ tests/integration/test_org_switching.py | 16 ++++ tests/integration/test_rag_integration.py | 52 +++++++++++ .../test_service_to_service_oauth.py | 31 +++++++ tests/integration/test_services_health.py | 77 +++++++++++++++++ tests/integration/test_tes_integration.py | 41 +++++++++ tests/test_backup_health_check.py | 21 +++++ tests/test_backup_mysql.py | 65 ++++++++++++++ tests/test_canonical_compose_path.py | 12 +++ tests/test_check_env.py | 8 ++ tests/test_compose_network_exposure.py | 20 +++-- tests/test_compose_release_config.py | 8 ++ tests/test_integration_credentials.py | 10 +++ tests/test_lib_alert.py | 20 +++++ tests/test_lib_env.py | 21 +++++ tests/test_lims_debug_config.py | 5 ++ tests/test_mysql_recovery_drill.py | 13 +++ tests/test_safe_compose_diagnostics.py | 12 +++ tests/test_service_catalog_drift.py | 9 ++ tests/test_verify_mysql_backup_restore.py | 26 ++++++ tests/unit/test_license_server.py | 38 ++++++++ 30 files changed, 794 insertions(+), 6 deletions(-) diff --git a/docker-socket-proxy/tests/test_policy.py b/docker-socket-proxy/tests/test_policy.py index 32a8ec8f..42b30556 100644 --- a/docker-socket-proxy/tests/test_policy.py +++ b/docker-socket-proxy/tests/test_policy.py @@ -8,6 +8,9 @@ Each "must still work" test encodes what this codebase's real docker.sock consumers actually send (grepped and confirmed against omnibioai/plugin_executor/ml_utils.py's real docker run invocation). + +Developer: + Manish Kumar """ import json import sys @@ -40,19 +43,30 @@ def _body(d: dict) -> bytes: # --------------------------------------------------------------------------- class TestEndpointAllowlist: + """Endpoint allowlist: container/image lifecycle endpoints are allowed; exec, swarm, + secrets, network and volume management, plugins, build, nodes and any unknown + endpoint are denied.""" def test_containers_json_allowed(self): + """GET /containers/json (container listing) is on the endpoint allowlist.""" assert check_endpoint("GET", "/containers/json").allowed def test_containers_create_allowed(self): + """POST /containers/create is allowed at the endpoint layer; its body is + validated separately.""" assert check_endpoint("POST", "/containers/create").allowed def test_versioned_path_allowed(self): + """A version-prefixed path such as /v1.43/containers/json is normalized and + allowed.""" assert check_endpoint("GET", "/v1.43/containers/json").allowed def test_images_pull_allowed(self): + """POST /images/create (image pull) is allowed.""" assert check_endpoint("POST", "/images/create").allowed def test_container_start_stop_wait_logs_attach_allowed(self): + """Start, stop, wait, logs, attach and DELETE on a specific container id are all + allowed.""" for method, path in [ ("POST", "/containers/abc123/start"), ("POST", "/containers/abc123/stop"), @@ -64,34 +78,43 @@ def test_container_start_stop_wait_logs_attach_allowed(self): assert check_endpoint(method, path).allowed, f"{method} {path} should be allowed" def test_exec_blocked(self): + """POST /containers/{id}/exec is denied.""" d = check_endpoint("POST", "/containers/abc123/exec") assert not d.allowed def test_exec_start_blocked(self): + """POST /exec/{id}/start is denied.""" d = check_endpoint("POST", "/exec/abc123/start") assert not d.allowed def test_swarm_blocked(self): + """The swarm API (/swarm/init) is denied.""" assert not check_endpoint("POST", "/swarm/init").allowed def test_secrets_blocked(self): + """Both listing and creating secrets through the Docker API are denied.""" assert not check_endpoint("GET", "/secrets").allowed assert not check_endpoint("POST", "/secrets/create").allowed def test_networks_management_blocked(self): + """Creating and deleting networks through the Docker API are both denied.""" assert not check_endpoint("POST", "/networks/create").allowed assert not check_endpoint("DELETE", "/networks/abc").allowed def test_volumes_management_blocked(self): + """Creating volumes through /volumes/create is denied.""" assert not check_endpoint("POST", "/volumes/create").allowed def test_plugins_blocked(self): + """GET /plugins is denied.""" assert not check_endpoint("GET", "/plugins").allowed def test_build_blocked(self): + """POST /build is denied.""" assert not check_endpoint("POST", "/build").allowed def test_nodes_blocked(self): + """GET /nodes (swarm node API) is denied.""" assert not check_endpoint("GET", "/nodes").allowed def test_unknown_endpoint_defaults_denied(self): @@ -100,6 +123,8 @@ def test_unknown_endpoint_defaults_denied(self): assert not check_endpoint("POST", "/some/brand/new/v2/endpoint").allowed def test_deny_reason_is_legible(self): + """A denied /swarm/init decision carries a reason that mentions swarm or the + allowlist.""" d = check_endpoint("POST", "/swarm/init") assert "swarm" in d.reason.lower() or "not on the allowlist" in d.reason.lower() @@ -109,21 +134,27 @@ def test_deny_reason_is_legible(self): # --------------------------------------------------------------------------- class TestPrivilegedBlocked: + """HostConfig.Privileged validation on /containers/create bodies.""" def test_privileged_true_blocked(self): + """HostConfig.Privileged=true is denied and the reason names 'privileged'.""" d = check_create_body(_body({"Image": "alpine", "HostConfig": {"Privileged": True}}), POLICY) assert not d.allowed assert "privileged" in d.reason.lower() def test_privileged_false_allowed(self): + """HostConfig.Privileged=false is allowed.""" d = check_create_body(_body({"Image": "alpine", "HostConfig": {"Privileged": False}}), POLICY) assert d.allowed def test_privileged_absent_allowed(self): + """A HostConfig with no Privileged key is allowed.""" d = check_create_body(_body({"Image": "alpine", "HostConfig": {}}), POLICY) assert d.allowed class TestHostBindMountBlocked: + """Bind and Mounts source validation: a source must be absolute and stay under an + allowed prefix after normalization, otherwise the create request is denied.""" def test_dotdot_traversal_out_of_allowed_prefix_blocked(self): """Caught in review, before merge: a naive string-prefix check lets '/app/work/../../../etc' through because it literally @@ -162,6 +193,7 @@ def test_dotdot_traversal_that_stays_inside_prefix_allowed(self): assert d.allowed def test_relative_bind_source_blocked(self): + """A relative bind source such as 'relative/path:/x' is denied.""" d = check_create_body( _body({"Image": "alpine", "HostConfig": {"Binds": ["relative/path:/x"]}}), POLICY ) @@ -176,12 +208,14 @@ def test_root_bind_mount_blocked(self): assert "outside the allowed prefixes" in d.reason def test_etc_bind_mount_blocked(self): + """Bind-mounting the host /etc is denied.""" d = check_create_body( _body({"Image": "alpine", "HostConfig": {"Binds": ["/etc:/hostetc"]}}), POLICY ) assert not d.allowed def test_home_bind_mount_blocked(self): + """Bind-mounting the host /home is denied.""" d = check_create_body( _body({"Image": "alpine", "HostConfig": {"Binds": ["/home:/hosthome"]}}), POLICY ) @@ -218,12 +252,16 @@ def test_allowed_prefix_bind_mount_allowed(self): assert d.allowed def test_exact_prefix_dir_itself_allowed(self): + """Binding the allowed prefix directory itself (/app/work) is allowed, not only + its subdirectories.""" d = check_create_body( _body({"Image": "alpine", "HostConfig": {"Binds": ["/app/work:/work"]}}), POLICY ) assert d.allowed def test_mounts_form_validated_same_as_binds(self): + """A HostConfig.Mounts bind entry with Source '/' is denied, the same as the + Binds form.""" d = check_create_body( _body({"Image": "alpine", "HostConfig": { "Mounts": [{"Type": "bind", "Source": "/", "Target": "/hostroot"}] @@ -232,6 +270,8 @@ def test_mounts_form_validated_same_as_binds(self): assert not d.allowed def test_mounts_form_allowed_prefix(self): + """A HostConfig.Mounts bind entry whose Source is under an allowed prefix is + allowed.""" d = check_create_body( _body({"Image": "alpine", "HostConfig": { "Mounts": [{"Type": "bind", "Source": "/app/work/run-1", "Target": "/work"}] @@ -240,6 +280,7 @@ def test_mounts_form_allowed_prefix(self): assert d.allowed def test_no_binds_no_mounts_allowed(self): + """A body with neither Binds nor Mounts is allowed.""" d = check_create_body(_body({"Image": "alpine", "HostConfig": {}}), POLICY) assert d.allowed @@ -254,6 +295,8 @@ class TestNamedVolumeBindsAllowlist: volumes are denied by default).""" def test_allowlisted_named_volume_allowed(self): + """A read-only bind of the allowlisted docker-proxy-socket named volume is + allowed.""" d = check_create_body( _body({"Image": "alpine", "HostConfig": { "Binds": ["docker-proxy-socket:/var/run/proxy-socket:ro"] @@ -327,6 +370,8 @@ def test_no_mode_segment_at_all_blocked(self): assert "must be mounted read-only" in d.reason def test_explicit_rw_blocked(self): + """An explicit :rw mode on the allowlisted named volume is denied with a + must-be-mounted-read-only reason.""" d = check_create_body( _body({"Image": "alpine", "HostConfig": { "Binds": ["docker-proxy-socket:/var/run/proxy-socket:rw"] @@ -346,6 +391,7 @@ def test_ro_combined_with_rw_blocked(self): assert not d.allowed def test_plain_ro_allowed(self): + """An explicit :ro mode on the allowlisted named volume is allowed.""" d = check_create_body( _body({"Image": "alpine", "HostConfig": { "Binds": ["docker-proxy-socket:/var/run/proxy-socket:ro"] @@ -383,6 +429,9 @@ class TestVolumeTypeMountBlocked: ever sends anyway.""" def test_volume_type_with_bind_driveropts_blocked_even_with_decoy_source(self): + """A Type=volume mount is denied even when its Source is a decoy under an + allowed prefix and the real host device is set through DriverConfig options; the + reason mentions the mount type.""" d = check_create_body( _body({"Image": "alpine", "HostConfig": {"Mounts": [{ "Type": "volume", @@ -397,6 +446,7 @@ def test_volume_type_with_bind_driveropts_blocked_even_with_decoy_source(self): assert "type" in d.reason.lower() def test_tmpfs_type_blocked(self): + """A Type=tmpfs mount is denied, since only bind mounts are accepted.""" d = check_create_body( _body({"Image": "alpine", "HostConfig": {"Mounts": [ {"Type": "tmpfs", "Target": "/tmp/x"} @@ -405,6 +455,7 @@ def test_tmpfs_type_blocked(self): assert not d.allowed def test_bind_type_still_allowed(self): + """A Type=bind mount whose Source is under an allowed prefix is still allowed.""" d = check_create_body( _body({"Image": "alpine", "HostConfig": {"Mounts": [ {"Type": "bind", "Source": "/app/work/run-1", "Target": "/work"} @@ -413,6 +464,8 @@ def test_bind_type_still_allowed(self): assert d.allowed def test_missing_type_defaults_to_bind_and_is_allowed(self): + """A Mounts entry with no Type is treated as a bind mount and allowed when its + Source is under an allowed prefix.""" d = check_create_body( _body({"Image": "alpine", "HostConfig": {"Mounts": [ {"Source": "/app/work/run-1", "Target": "/work"} @@ -422,7 +475,11 @@ def test_missing_type_defaults_to_bind_and_is_allowed(self): class TestDevicesSecurityOptUsernsBlocked: + """Denies host Devices, weakening SecurityOpt values (seccomp or apparmor + unconfined, no-new-privileges=false) and UsernsMode=host, while benign settings stay + allowed.""" def test_devices_blocked(self): + """Passing a host device (/dev/sda) through HostConfig.Devices is denied.""" d = check_create_body( _body({"Image": "alpine", "HostConfig": { "Devices": [{"PathOnHost": "/dev/sda", "PathInContainer": "/dev/sda", "CgroupPermissions": "rwm"}] @@ -431,34 +488,40 @@ def test_devices_blocked(self): assert not d.allowed def test_no_devices_allowed(self): + """An empty Devices list is allowed.""" d = check_create_body(_body({"Image": "alpine", "HostConfig": {"Devices": []}}), POLICY) assert d.allowed def test_seccomp_unconfined_blocked(self): + """SecurityOpt seccomp=unconfined is denied.""" d = check_create_body( _body({"Image": "alpine", "HostConfig": {"SecurityOpt": ["seccomp=unconfined"]}}), POLICY ) assert not d.allowed def test_apparmor_unconfined_blocked(self): + """SecurityOpt apparmor=unconfined is denied.""" d = check_create_body( _body({"Image": "alpine", "HostConfig": {"SecurityOpt": ["apparmor=unconfined"]}}), POLICY ) assert not d.allowed def test_no_new_privileges_false_blocked(self): + """SecurityOpt no-new-privileges=false is denied.""" d = check_create_body( _body({"Image": "alpine", "HostConfig": {"SecurityOpt": ["no-new-privileges=false"]}}), POLICY ) assert not d.allowed def test_benign_security_opt_allowed(self): + """SecurityOpt no-new-privileges=true is allowed.""" d = check_create_body( _body({"Image": "alpine", "HostConfig": {"SecurityOpt": ["no-new-privileges=true"]}}), POLICY ) assert d.allowed def test_userns_mode_host_blocked(self): + """UsernsMode=host is denied.""" d = check_create_body( _body({"Image": "alpine", "HostConfig": {"UsernsMode": "host"}}), POLICY ) @@ -466,7 +529,10 @@ def test_userns_mode_host_blocked(self): class TestCapAddBlocked: + """Linux capability grants through HostConfig.CapAdd are denied; an empty list is + allowed.""" def test_cap_add_sys_admin_blocked(self): + """CapAdd SYS_ADMIN is denied and the reason mentions capadd.""" d = check_create_body( _body({"Image": "alpine", "HostConfig": {"CapAdd": ["SYS_ADMIN"]}}), POLICY ) @@ -474,12 +540,16 @@ def test_cap_add_sys_admin_blocked(self): assert "capadd" in d.reason.lower() def test_empty_cap_add_allowed(self): + """An empty CapAdd list is allowed.""" d = check_create_body(_body({"Image": "alpine", "HostConfig": {"CapAdd": []}}), POLICY) assert d.allowed class TestHostNamespacesBlocked: + """Host namespace sharing through NetworkMode, PidMode and IpcMode set to host is + denied; bridge networking stays allowed.""" def test_network_mode_host_blocked(self): + """NetworkMode=host is denied and the reason names networkmode.""" d = check_create_body( _body({"Image": "alpine", "HostConfig": {"NetworkMode": "host"}}), POLICY ) @@ -487,30 +557,39 @@ def test_network_mode_host_blocked(self): assert "networkmode" in d.reason.lower() def test_network_mode_bridge_allowed(self): + """NetworkMode=bridge is allowed.""" d = check_create_body( _body({"Image": "alpine", "HostConfig": {"NetworkMode": "bridge"}}), POLICY ) assert d.allowed def test_pid_mode_host_blocked(self): + """PidMode=host is denied.""" d = check_create_body(_body({"Image": "alpine", "HostConfig": {"PidMode": "host"}}), POLICY) assert not d.allowed def test_ipc_mode_host_blocked(self): + """IpcMode=host is denied.""" d = check_create_body(_body({"Image": "alpine", "HostConfig": {"IpcMode": "host"}}), POLICY) assert not d.allowed class TestMalformedBody: + """Malformed /containers/create bodies: invalid JSON, non-object JSON, a non-object + HostConfig and an unparseable Binds entry are denied; an empty body is allowed.""" def test_invalid_json_blocked(self): + """A body that is not valid JSON is denied.""" d = check_create_body(b"{not valid json", POLICY) assert not d.allowed def test_non_object_json_blocked(self): + """A JSON body that is not an object (a list) is denied.""" d = check_create_body(b"[1, 2, 3]", POLICY) assert not d.allowed def test_empty_body_allowed(self): + """An empty create body is allowed; it fails open here by design, with the + endpoint allowlist and the daemon's own validation as the backstop.""" # In practice /containers/create always has a body from a real # client; an empty body isn't itself a way to request anything # dangerous, so this fails open here rather than blocking @@ -520,10 +599,12 @@ def test_empty_body_allowed(self): assert d.allowed def test_non_dict_host_config_blocked(self): + """A HostConfig value that is not an object is denied.""" d = check_create_body(_body({"Image": "alpine", "HostConfig": "not-a-dict"}), POLICY) assert not d.allowed def test_unparseable_bind_entry_blocked(self): + """A Binds entry with no source:target colon is denied.""" d = check_create_body( _body({"Image": "alpine", "HostConfig": {"Binds": ["no-colon-here"]}}), POLICY ) @@ -535,6 +616,8 @@ def test_unparseable_bind_entry_blocked(self): # --------------------------------------------------------------------------- class TestEvaluateRequest: + """End-to-end evaluate_request(): the endpoint allowlist and create-body validation + applied together.""" def test_realistic_ml_utils_create_call_allowed(self): """Mirrors omnibioai/plugin_executor/ml_utils.py's real `docker run --rm --gpus all -v run_dir:/work -v cache_dir:/root/.cache/torch @@ -555,6 +638,8 @@ def test_realistic_ml_utils_create_call_allowed(self): assert d.allowed def test_privileged_escape_attempt_blocked_end_to_end(self): + """A /containers/create combining Privileged=true with a '/' bind mount is + denied through evaluate_request.""" body = _body({ "Image": "alpine", "Cmd": ["sh"], @@ -564,6 +649,7 @@ def test_privileged_escape_attempt_blocked_end_to_end(self): assert not d.allowed def test_exec_into_running_container_blocked_end_to_end(self): + """POST /v1.43/containers/{id}/exec is denied through evaluate_request.""" d = evaluate_request("POST", "/v1.43/containers/some-id/exec", b"", POLICY) assert not d.allowed diff --git a/docker-socket-proxy/tests/test_proxy_connection_handling.py b/docker-socket-proxy/tests/test_proxy_connection_handling.py index c3677762..e3ac7c5e 100644 --- a/docker-socket-proxy/tests/test_proxy_connection_handling.py +++ b/docker-socket-proxy/tests/test_proxy_connection_handling.py @@ -24,6 +24,9 @@ TestConnectionReuseDoesNotCauseIntermittentFailures for that live, real-daemon reproduction; these tests pin down the pure header-framing fix in isolation, no daemon needed. + +Developer: + Manish Kumar """ from __future__ import annotations @@ -36,11 +39,15 @@ def _raw(status_line: str, *header_lines: str) -> bytes: + """Builds raw response header bytes (status line plus header lines, blank-line + terminated, ISO-8859-1) as input for _force_connection_close.""" text = "\r\n".join([status_line, *header_lines, "", ""]) return text.encode("iso-8859-1") class TestForceConnectionClose: + """_force_connection_close() rewrites an upstream response header block so it always + carries exactly one Connection: close header and leaves everything else intact.""" def test_adds_close_when_no_connection_header_present(self): """The exact real-world case that caused the bug: the real daemon's /_ping response carries no Connection header at all.""" @@ -52,6 +59,8 @@ def test_adds_close_when_no_connection_header_present(self): assert text.count("Connection:") == 1 def test_overrides_existing_keep_alive_value(self): + """An upstream Connection: keep-alive header is replaced by a single Connection: + close.""" raw = _raw("HTTP/1.1 200 OK", "Content-Length: 0", "Connection: keep-alive") out = _force_connection_close(raw).decode("iso-8859-1") assert "Connection: close" in out @@ -59,6 +68,8 @@ def test_overrides_existing_keep_alive_value(self): assert out.count("Connection:") == 1 def test_case_insensitive_header_name_match(self): + """The Connection header name is matched case-insensitively, so 'CONNECTION: + Keep-Alive' is replaced and exactly one Connection header remains.""" raw = _raw("HTTP/1.1 200 OK", "content-length: 0", "CONNECTION: Keep-Alive") out = _force_connection_close(raw).decode("iso-8859-1") assert "Connection: close" in out @@ -67,6 +78,8 @@ def test_case_insensitive_header_name_match(self): assert sum(1 for line in out.split("\r\n") if line.lower().startswith("connection:")) == 1 def test_preserves_status_line_and_other_headers(self): + """The status line and unrelated headers are kept unchanged when Connection: + close is added.""" raw = _raw( "HTTP/1.1 404 Not Found", "Content-Type: application/json", @@ -82,6 +95,8 @@ def test_preserves_status_line_and_other_headers(self): assert "Connection: close" in lines def test_no_headers_besides_status_line(self): + """A response consisting only of a status line still gets a Connection: close + header.""" raw = _raw("HTTP/1.1 204 No Content") out = _force_connection_close(raw).decode("iso-8859-1") lines = out.split("\r\n") @@ -89,6 +104,8 @@ def test_no_headers_besides_status_line(self): assert "Connection: close" in lines def test_output_ends_with_blank_line_terminator(self): + """The rewritten header block still ends with a single CRLF CRLF terminator and + nothing follows it.""" raw = _raw("HTTP/1.1 200 OK", "Content-Length: 0") out = _force_connection_close(raw) assert out.endswith(b"\r\n\r\n") diff --git a/docker-socket-proxy/tests/test_proxy_live_integration.py b/docker-socket-proxy/tests/test_proxy_live_integration.py index 15c2737e..bcd82409 100644 --- a/docker-socket-proxy/tests/test_proxy_live_integration.py +++ b/docker-socket-proxy/tests/test_proxy_live_integration.py @@ -12,6 +12,9 @@ environment -- this is an opt-in live test, matching the pattern established elsewhere in this codebase for tests that need a real Docker daemon (e.g. pdf_report_builder's opt-in live E2E). + +Developer: + Manish Kumar """ from __future__ import annotations @@ -87,6 +90,8 @@ def proxy_env(tmp_path_factory): def _docker(env, *args, timeout=30): + """Runs the docker CLI with the given environment (whose DOCKER_HOST points at the + proxy socket) and captures its output.""" return subprocess.run( [DOCKER_BIN, *args], env=env, @@ -107,11 +112,14 @@ class TestLegitimateTrafficWorksEndToEnd: without breaking it.""" def test_docker_ps(self, proxy_env): + """docker ps through the proxy exits 0.""" env, _ = proxy_env result = _docker(env, "ps") assert result.returncode == 0, result.stderr def test_run_with_allowed_bind_mount_full_round_trip(self, proxy_env): + """A docker run bind-mounting the allowed workdir succeeds, and the file the + container wrote is visible in the host workdir.""" env, workdir = proxy_env name = _unique_container_name() result = _docker( @@ -132,6 +140,8 @@ def test_pull_already_present_image(self, proxy_env): assert result.returncode == 0, result.stderr def test_logs_on_a_running_container(self, proxy_env): + """docker logs on a running proxied container returns its output; the container + is force-removed afterwards.""" env, workdir = proxy_env name = _unique_container_name() try: @@ -155,6 +165,8 @@ class TestAdversarialRequestsAreBlocked: (403-style denial) -- never merely fail for some other reason.""" def test_privileged_blocked(self, proxy_env): + """docker run --privileged is rejected by the proxy with a denial message that + names privileged.""" env, workdir = proxy_env result = _docker( env, "run", "--rm", "--privileged", @@ -165,12 +177,15 @@ def test_privileged_blocked(self, proxy_env): assert "privileged" in result.stderr.lower() def test_root_bind_mount_blocked(self, proxy_env): + """docker run -v /:/hostroot is rejected by the proxy with a denial message.""" env, _ = proxy_env result = _docker(env, "run", "--rm", "-v", "/:/hostroot", "alpine:latest", "true") assert result.returncode != 0 assert "denied" in result.stderr.lower() def test_bind_mount_outside_allowed_prefix_blocked(self, proxy_env): + """A bind mount of /tmp, outside the allowed workdir prefix, is rejected by the + proxy with a denial message.""" env, _ = proxy_env result = _docker(env, "run", "--rm", "-v", "/tmp:/hosttmp", "alpine:latest", "true") assert result.returncode != 0 @@ -190,6 +205,8 @@ def test_docker_sock_remount_blocked(self, proxy_env): assert "denied" in result.stderr.lower() def test_cap_add_blocked(self, proxy_env): + """docker run --cap-add SYS_ADMIN is rejected by the proxy with a denial message + that names capadd.""" env, workdir = proxy_env result = _docker( env, "run", "--rm", "--cap-add", "SYS_ADMIN", @@ -200,6 +217,8 @@ def test_cap_add_blocked(self, proxy_env): assert "capadd" in result.stderr.lower() def test_network_host_blocked(self, proxy_env): + """docker run --network host is rejected by the proxy with a denial message that + names networkmode.""" env, workdir = proxy_env result = _docker( env, "run", "--rm", "--network", "host", @@ -210,6 +229,8 @@ def test_network_host_blocked(self, proxy_env): assert "networkmode" in result.stderr.lower() def test_exec_into_running_container_blocked(self, proxy_env): + """docker exec into a running proxied container is rejected with a denial + message citing the allowlist; the container is force-removed afterwards.""" env, workdir = proxy_env name = _unique_container_name() try: @@ -251,6 +272,8 @@ class TestConnectionReuseDoesNotCauseIntermittentFailures: against the real published (pre-fix) image.""" def test_many_sequential_legitimate_runs_all_succeed(self, proxy_env): + """Fifteen back-to-back allowed docker runs all succeed with no reset, + broken-pipe or closed-idle-connection errors, and each writes its output file.""" env, workdir = proxy_env for i in range(15): result = _docker( diff --git a/tests/integration/conftest.py b/tests/integration/conftest.py index 9f66cf10..1c024e74 100644 --- a/tests/integration/conftest.py +++ b/tests/integration/conftest.py @@ -3,6 +3,9 @@ Services are accessed through the nginx router at http://localhost (port 80) and also directly via their native ports where noted. + +Developer: + Manish Kumar """ import os @@ -99,6 +102,9 @@ def _login_or_skip() -> dict: @pytest.fixture(scope="session") def auth_tokens(): + """Session-scoped login to the auth service with the + OMNIBIOAI_AUTH_EMAIL/OMNIBIOAI_AUTH_PASSWORD credentials; skips when credentials are + not configured or the service is unreachable.""" return _login_or_skip() @@ -144,6 +150,9 @@ def _lims_login_or_skip() -> requests.Session: @pytest.fixture(scope="session") def lims_session(): + """Session-scoped requests.Session authenticated against LIMS with the + OMNIBIOAI_LIMS_USER/OMNIBIOAI_LIMS_PASSWORD credentials; skips when LIMS is + unreachable.""" return _lims_login_or_skip() @@ -151,4 +160,6 @@ def lims_session(): @pytest.fixture(scope="session") def rag_headers(): + """Authorization header built from RAGBIO_API_KEY; the key is an empty string when + the environment variable is not set.""" return {"Authorization": f"Bearer {RAGBIO_API_KEY}"} diff --git a/tests/integration/test_auth_integration.py b/tests/integration/test_auth_integration.py index 08671d50..44776624 100644 --- a/tests/integration/test_auth_integration.py +++ b/tests/integration/test_auth_integration.py @@ -19,6 +19,9 @@ Tests work around this by caching the one successful login result per module and injecting session-level tokens from conftest for fixtures that previously called login on every test. + +Developer: + Manish Kumar """ import pytest @@ -64,11 +67,14 @@ def _get(path: str) -> requests.Response: # ── Health ──────────────────────────────────────────────────────────────────── class TestAuthHealth: + """Liveness probe of the auth service: GET /health responds 200 with status ok.""" def test_health_returns_200(self): + """GET /health on the auth service returns HTTP 200.""" r = _get("/health") assert r.status_code == 200 def test_health_body_status_ok(self): + """GET /health returns a body whose status is ok.""" r = _get("/health") assert r.json()["status"] == "ok" @@ -78,37 +84,49 @@ def test_health_body_status_ok(self): # successful login cached in _obtain_auth_tokens(). class TestAuthLogin: + """POST /auth/login response shape and rejection cases, reusing the one successful + login cached per module; skips when the service could not issue tokens.""" @pytest.fixture(autouse=True) def _login_data(self): + """Autouse fixture: exposes the cached login tokens on self, and skips the test + when none could be obtained.""" tokens = _obtain_auth_tokens() if tokens is None: pytest.skip("Auth service could not issue tokens (duplicate token race)") self._tokens = tokens def test_login_with_valid_credentials_returns_200(self): + """A login with the configured credentials succeeded, so the cached tokens from + its 200 response are present.""" # Tokens were obtained via a 200 response; if we got here, login worked. assert self._tokens is not None def test_login_returns_access_token(self): + """The login response includes a non-empty access_token.""" assert "access_token" in self._tokens assert self._tokens["access_token"] def test_login_returns_refresh_token(self): + """The login response includes a non-empty refresh_token.""" assert "refresh_token" in self._tokens assert self._tokens["refresh_token"] def test_login_token_type_is_bearer(self): + """The login response reports token_type bearer.""" assert self._tokens.get("token_type") == "bearer" def test_login_with_wrong_password_returns_401(self): + """Logging in with a wrong password returns 401.""" r = _post("/auth/login", {"email": AUTH_ADMIN_EMAIL, "password": "wrong-password"}) assert r.status_code == 401 def test_login_with_unknown_email_returns_401(self): + """Logging in with an unknown email returns 401.""" r = _post("/auth/login", {"email": "nobody@nowhere.com", "password": "x"}) assert r.status_code == 401 def test_login_missing_fields_returns_422(self): + """Logging in with an empty JSON body returns 422.""" r = _post("/auth/login", {}) assert r.status_code == 422 @@ -116,34 +134,44 @@ def test_login_missing_fields_returns_422(self): # ── Token validation ────────────────────────────────────────────────────────── class TestAuthValidate: + """POST /auth/validate for valid, garbage and empty tokens, using the session-level + tokens from conftest.""" @pytest.fixture(autouse=True) def _tokens(self, access_token, refresh_token): + """Autouse fixture: injects the session-level access and refresh tokens onto + self so /auth/login is not called again.""" # Inject session-level tokens from conftest — avoids calling /auth/login again. self.access_token = access_token self.refresh_token = refresh_token def test_validate_valid_token_returns_200(self): + """POST /auth/validate for a valid access token returns HTTP 200.""" r = _post("/auth/validate", {"token": self.access_token}) assert r.status_code == 200 def test_validate_valid_token_returns_true(self): + """POST /auth/validate for a valid access token reports valid=true.""" r = _post("/auth/validate", {"token": self.access_token}) assert r.json()["valid"] is True def test_validate_valid_token_contains_user_id(self): + """The validate response for a valid token includes a user_id.""" r = _post("/auth/validate", {"token": self.access_token}) assert "user_id" in r.json() def test_validate_valid_token_contains_email(self): + """The validate response email matches the configured admin email.""" r = _post("/auth/validate", {"token": self.access_token}) assert r.json()["email"] == AUTH_ADMIN_EMAIL def test_validate_garbage_token_returns_false(self): + """A garbage token returns HTTP 200 with valid=false.""" r = _post("/auth/validate", {"token": "not.a.real.jwt"}) assert r.status_code == 200 assert r.json()["valid"] is False def test_validate_empty_token_returns_false(self): + """An empty token is reported as valid=false.""" r = _post("/auth/validate", {"token": ""}) assert r.json()["valid"] is False @@ -151,28 +179,36 @@ def test_validate_empty_token_returns_false(self): # ── Refresh ─────────────────────────────────────────────────────────────────── class TestAuthRefresh: + """POST /auth/refresh with valid and invalid refresh tokens, using the session-level + tokens from conftest.""" @pytest.fixture(autouse=True) def _tokens(self, access_token, refresh_token): + """Autouse fixture: injects the session-level access and refresh tokens onto + self so /auth/login is not called again.""" # Inject session-level tokens from conftest — avoids calling /auth/login again. self.access_token = access_token self.refresh_token = refresh_token def test_refresh_valid_token_returns_200(self): + """POST /auth/refresh with a valid refresh token returns HTTP 200.""" r = _post("/auth/refresh", {"refresh_token": self.refresh_token}) assert r.status_code == 200 def test_refresh_returns_new_access_token(self): + """The refresh response includes a non-empty access_token.""" r = _post("/auth/refresh", {"refresh_token": self.refresh_token}) assert "access_token" in r.json() assert r.json()["access_token"] def test_new_access_token_is_valid(self): + """The access token issued by a refresh validates as valid=true.""" r = _post("/auth/refresh", {"refresh_token": self.refresh_token}) new_token = r.json()["access_token"] validate_r = _post("/auth/validate", {"token": new_token}) assert validate_r.json()["valid"] is True def test_refresh_invalid_token_returns_401(self): + """POST /auth/refresh with an invalid refresh token returns 401.""" r = _post("/auth/refresh", {"refresh_token": "bad-token"}) assert r.status_code == 401 @@ -183,29 +219,40 @@ def test_refresh_invalid_token_returns_401(self): # already-revoked state gracefully. class TestAuthLogout: + """POST /auth/logout revokes the module-cached refresh token, and a revoked token + can no longer be refreshed. These tests deliberately revoke tokens that later tests + must tolerate.""" @pytest.fixture(autouse=True) def _cached(self): + """Autouse fixture: exposes the cached refresh token on self, and skips the test + when no tokens are available.""" tokens = _obtain_auth_tokens() if tokens is None: pytest.skip("No auth tokens available") self._refresh_token = tokens["refresh_token"] def test_logout_returns_200(self): + """POST /auth/logout with the cached refresh token returns HTTP 200.""" r = _post("/auth/logout", {"refresh_token": self._refresh_token}) assert r.status_code == 200 def test_logout_returns_message(self): + """Logout returns 200 with a message field, even when the token was already + revoked.""" # May be called after token is already revoked — double-revoke still returns 200. r = _post("/auth/logout", {"refresh_token": self._refresh_token}) assert r.status_code == 200 assert "message" in r.json() def test_token_invalid_after_logout(self): + """After logout, refreshing with the same refresh token returns 401.""" _post("/auth/logout", {"refresh_token": self._refresh_token}) r = _post("/auth/refresh", {"refresh_token": self._refresh_token}) assert r.status_code == 401 def test_logout_invalid_token_returns_error(self): + """Logging out with an invalid token returns one of 200, 400, 401 or 422, so an + idempotent revoke is tolerated.""" # Some implementations return 200 for invalid tokens (idempotent revoke). r = _post("/auth/logout", {"refresh_token": "invalid-token"}) assert r.status_code in (200, 400, 401, 422) @@ -214,7 +261,12 @@ def test_logout_invalid_token_returns_error(self): # ── Full round-trip ─────────────────────────────────────────────────────────── class TestAuthFullFlow: + """End-to-end login, validate, refresh and logout sequence; the refresh step xfails + when an earlier logout test already revoked the token.""" def test_login_refresh_validate_logout(self, auth_tokens, access_token, refresh_token): + """Login, validate, refresh, validate the new token, log out, then confirm a + refresh after logout returns 401; xfails when the refresh token was already + revoked by an earlier logout test.""" # Login step: confirmed by the session-level auth_tokens fixture. assert auth_tokens.get("access_token"), "Login did not return access_token" diff --git a/tests/integration/test_auth_registration.py b/tests/integration/test_auth_registration.py index 43f7bf7c..f0e6dfe6 100644 --- a/tests/integration/test_auth_registration.py +++ b/tests/integration/test_auth_registration.py @@ -12,6 +12,9 @@ there is no email-verification step, so a freshly registered account can log in right away. Each test generates a unique email via uuid4 so repeat runs never collide with a previous run's leftover user row. + +Developer: + Manish Kumar """ import uuid @@ -32,15 +35,21 @@ def _post(path: str, body: dict) -> requests.Response: class TestRegisterNewUser: + """Successful POST /auth/register calls with unique emails, and the resulting + account's login behavior.""" def test_register_returns_200(self): + """Registering a new unique email with a valid password returns HTTP 200.""" r = _post("/auth/register", {"email": _unique_email(), "password": "S3curePass!1"}) assert r.status_code == 200 def test_register_returns_message(self): + """The register response body includes a message field.""" r = _post("/auth/register", {"email": _unique_email(), "password": "S3curePass!1"}) assert "message" in r.json() def test_registered_user_can_log_in_immediately(self): + """A freshly registered user can log in at once and receives both an access and + a refresh token.""" email = _unique_email() password = "S3curePass!1" @@ -53,6 +62,7 @@ def test_registered_user_can_log_in_immediately(self): assert "refresh_token" in login.json() def test_registered_user_wrong_password_rejected(self): + """A registered user logging in with a wrong password gets 401.""" email = _unique_email() _post("/auth/register", {"email": email, "password": "S3curePass!1"}) @@ -61,7 +71,9 @@ def test_registered_user_wrong_password_rejected(self): class TestRegisterDuplicate: + """Registering an email that already exists is rejected.""" def test_duplicate_email_returns_400(self): + """Registering the same email a second time returns 400.""" email = _unique_email() first = _post("/auth/register", {"email": email, "password": "S3curePass!1"}) assert first.status_code == 200 @@ -71,10 +83,13 @@ def test_duplicate_email_returns_400(self): class TestRegisterValidation: + """Request-schema validation on POST /auth/register.""" def test_missing_fields_returns_422(self): + """An empty register body returns 422.""" r = _post("/auth/register", {}) assert r.status_code == 422 def test_missing_password_returns_422(self): + """A register body with an email but no password returns 422.""" r = _post("/auth/register", {"email": _unique_email()}) assert r.status_code == 422 diff --git a/tests/integration/test_auth_revocation.py b/tests/integration/test_auth_revocation.py index 324a44ce..1d2e3169 100644 --- a/tests/integration/test_auth_revocation.py +++ b/tests/integration/test_auth_revocation.py @@ -18,6 +18,9 @@ of reusing conftest's session-scoped admin tokens, since this test deliberately revokes its own token and must not interfere with other test files sharing that session-scoped login. + +Developer: + Manish Kumar """ import uuid @@ -30,6 +33,8 @@ def _register_and_login() -> dict: + """Registers a throwaway user with a unique email against the auth service and logs + in, returning the token response; a failed setup step fails the test.""" email = f"itest-revoke-{uuid.uuid4().hex}@example.com" password = "S3curePass!1" @@ -50,12 +55,17 @@ def _list_orgs(access_token: str) -> requests.Response: class TestAccessTokenRevokedAfterLogout: + """Logout with the access token in the request body revokes it, and the refresh + token is rejected too; omitting the access token leaves it valid until it expires.""" def test_access_token_works_before_logout(self): + """A freshly issued access token can list orgs with HTTP 200 before logout.""" tokens = _register_and_login() r = _list_orgs(tokens["access_token"]) assert r.status_code == 200 def test_access_token_rejected_after_logout(self): + """After a logout that supplies both tokens, the same access token is rejected + with 401 on GET /orgs.""" tokens = _register_and_login() # Sanity: token is live before logout. @@ -72,6 +82,7 @@ def test_access_token_rejected_after_logout(self): assert r.status_code == 401 def test_refresh_token_also_rejected_after_logout(self): + """After logout, POST /auth/refresh with the refresh token returns 401.""" tokens = _register_and_login() requests.post( diff --git a/tests/integration/test_conftest_fixtures.py b/tests/integration/test_conftest_fixtures.py index 2b01ddc6..73eb996c 100644 --- a/tests/integration/test_conftest_fixtures.py +++ b/tests/integration/test_conftest_fixtures.py @@ -7,6 +7,9 @@ that back the auth_tokens / lims_session fixtures. Run: pytest tests/integration/test_conftest_fixtures.py -v + +Developer: + Manish Kumar """ import pytest @@ -26,36 +29,48 @@ # ── URL fixtures ────────────────────────────────────────────────────────────── def test_base_url_fixture(base_url): + """The base_url fixture returns conftest.BASE_URL.""" assert base_url == BASE_URL def test_auth_url_fixture(auth_url): + """The auth_url fixture returns conftest.AUTH_DIRECT_URL.""" assert auth_url == AUTH_DIRECT_URL def test_lims_url_fixture(lims_url): + """The lims_url fixture returns conftest.LIMS_DIRECT_URL.""" assert lims_url == LIMS_DIRECT_URL def test_tes_url_fixture(tes_url): + """The tes_url fixture returns conftest.TES_DIRECT_URL.""" assert tes_url == TES_DIRECT_URL def test_rag_url_fixture(rag_url): + """The rag_url fixture returns conftest.RAG_DIRECT_URL.""" assert rag_url == RAG_DIRECT_URL # ── http fixture ────────────────────────────────────────────────────────────── class TestHttpFixture: + """The shared http fixture: a requests.Session that applies the default timeout and + still accepts explicit request kwargs. Two tests call the live RAG /health endpoint.""" def test_is_a_session(self, http): + """The http fixture is a requests.Session instance.""" assert isinstance(http, requests.Session) def test_applies_default_timeout(self, http): + """A GET through the http fixture to the RAG service /health returns 200 with + the default-timeout wrapper in place.""" r = http.get(f"{RAG_DIRECT_URL}/health") assert r.status_code == 200 def test_can_still_pass_explicit_kwargs(self, http): + """http.get still accepts explicit headers and the RAG /health request returns + 200.""" r = http.get(f"{RAG_DIRECT_URL}/health", headers={"X-Test": "1"}) assert r.status_code == 200 @@ -63,10 +78,13 @@ def test_can_still_pass_explicit_kwargs(self, http): # ── auth_headers / rag_headers fixtures ────────────────────────────────────── def test_auth_headers_fixture(auth_headers, access_token): + """auth_headers is a Bearer Authorization header built from the access_token + fixture.""" assert auth_headers == {"Authorization": f"Bearer {access_token}"} def test_rag_headers_fixture(rag_headers): + """rag_headers is a Bearer Authorization header built from RAGBIO_API_KEY.""" assert rag_headers == {"Authorization": f"Bearer {RAGBIO_API_KEY}"} @@ -80,6 +98,8 @@ def raise_for_status(self): def test_login_or_skip_skips_when_auth_service_unreachable(monkeypatch): + """_login_or_skip turns a failing auth login response (stubbed so raise_for_status + raises) into a pytest skip rather than an error.""" monkeypatch.setattr( conftest.requests, "post", lambda *a, **k: _FailingResponse() ) @@ -88,6 +108,8 @@ def test_login_or_skip_skips_when_auth_service_unreachable(monkeypatch): def test_lims_login_or_skip_skips_when_lims_unreachable(monkeypatch): + """_lims_login_or_skip raises a pytest skip when the LIMS login POST raises a + connection error (stubbed).""" def _raise(*a, **k): raise requests.ConnectionError("no route to host") @@ -97,4 +119,6 @@ def _raise(*a, **k): def test_lims_session_fixture_authenticates(lims_session): + """The lims_session fixture yields a requests.Session after a login against the live + LIMS.""" assert isinstance(lims_session, requests.Session) diff --git a/tests/integration/test_cross_app_sso.py b/tests/integration/test_cross_app_sso.py index 65383ab6..4b17b13e 100644 --- a/tests/integration/test_cross_app_sso.py +++ b/tests/integration/test_cross_app_sso.py @@ -40,6 +40,9 @@ session-scoped admin login, since that fixture's fixed admin/admin credentials are shared (and sometimes already consumed) across every other test file in this suite. + +Developer: + Manish Kumar """ import uuid @@ -53,6 +56,8 @@ def _register_and_login() -> dict: + """Registers a throwaway user with a unique email in the central auth service and + logs in, returning the tokens plus the email.""" email = f"itest-crosssso-{uuid.uuid4().hex}@example.com" password = "S3curePass!1" @@ -72,7 +77,10 @@ def _register_and_login() -> dict: class TestCentralAuthTokenWorksAgainstLims: + """An access token issued by the central auth service is accepted directly by LIMS + at /api/auth/me/, with no LIMS login (the cross-application SSO contract).""" def test_central_access_token_authenticates_to_lims(self): + """GET /api/auth/me/ on LIMS with a central-auth access token returns HTTP 200.""" tokens = _register_and_login() r = requests.get( LIMS_ME_URL, @@ -82,6 +90,7 @@ def test_central_access_token_authenticates_to_lims(self): assert r.status_code == 200 def test_lims_resolves_the_same_email_from_the_central_token(self): + """LIMS's /api/auth/me/ reports the same email the central token was issued for.""" tokens = _register_and_login() r = requests.get( LIMS_ME_URL, @@ -105,11 +114,14 @@ def test_no_separate_lims_login_was_needed(self): class TestLimsRejectsWithoutAToken: + """LIMS /api/auth/me/ rejects requests that carry no valid token.""" def test_me_endpoint_requires_authentication(self): + """GET /api/auth/me/ with no Authorization header returns 401.""" r = requests.get(LIMS_ME_URL, headers=_TRUSTED_PROXY_HEADERS, timeout=TIMEOUT) assert r.status_code == 401 def test_me_endpoint_rejects_garbage_bearer_token(self): + """A garbage bearer token on /api/auth/me/ returns 401.""" r = requests.get( LIMS_ME_URL, headers={**_TRUSTED_PROXY_HEADERS, "Authorization": "Bearer not.a.real.jwt"}, diff --git a/tests/integration/test_lims_integration.py b/tests/integration/test_lims_integration.py index e815346d..09989c78 100644 --- a/tests/integration/test_lims_integration.py +++ b/tests/integration/test_lims_integration.py @@ -34,6 +34,9 @@ 443 here. This reproduces exactly what the real router already sets on every request LIMS actually receives; it is not bypassing a security check. + +Developer: + Manish Kumar """ import pytest @@ -103,11 +106,14 @@ def _obtain_tokens() -> dict: # ── Health probes ───────────────────────────────────────────────────────────── class TestLimsHealth: + """LIMS liveness (/healthz) and readiness (/readyz) probes respond 200.""" def test_healthz_returns_200(self): + """GET /healthz returns HTTP 200.""" r = _get("/healthz") assert r.status_code == 200 def test_readyz_returns_200(self): + """GET /readyz returns HTTP 200.""" r = _get("/readyz") assert r.status_code == 200 @@ -115,6 +121,8 @@ def test_readyz_returns_200(self): # ── JWT authentication ──────────────────────────────────────────────────────── class TestLimsAuth: + """POST /api/token/ delivers tokens as cookies and rejects wrong, unknown and + missing credentials; the login is cached because LIMS rate-limits the endpoint.""" @pytest.fixture(autouse=True) def _tokens(self): # Use cached login to avoid rate-limiting the /api/token/ endpoint. @@ -123,26 +131,35 @@ def _tokens(self): self._refresh = data["refresh"] def test_obtain_token_returns_200(self): + """The cached LIMS login succeeded and set a non-empty access_token cookie.""" # Verified by the fact that _obtain_tokens() succeeded (raise_for_status called). assert self._access, "access_token cookie was not set after successful login" def test_obtain_token_returns_access_field(self): + """The login delivers a non-empty access_token cookie; tokens are not in the + JSON body.""" # Tokens are in Set-Cookie headers, not in the JSON body. assert self._access, "access_token cookie not set" def test_obtain_token_returns_refresh_field(self): + """The login delivers a non-empty refresh_token cookie.""" assert self._refresh, "refresh_token cookie not set" def test_wrong_credentials_returns_401(self): + """A wrong password on /api/token/ returns 401, or 429 when the rate limiter + answers first.""" # 429 is also acceptable: rate limiter fires before credentials are checked. r = _post("/api/token/", {"username": LIMS_USERNAME, "password": "wrongpassword"}) assert r.status_code in (401, 429) def test_unknown_user_returns_401(self): + """An unknown username on /api/token/ returns 401, or 429 when the rate limiter + answers first.""" r = _post("/api/token/", {"username": "nobody", "password": "x"}) assert r.status_code in (401, 429) def test_missing_credentials_returns_400_or_422(self): + """An empty body on /api/token/ returns one of 400, 401, 422 or 429.""" r = _post("/api/token/", {}) assert r.status_code in (400, 401, 422, 429) @@ -152,6 +169,8 @@ def test_missing_credentials_returns_400_or_422(self): # Use a requests.Session so the cookie is sent automatically. class TestLimsTokenRefresh: + """POST /api/token/refresh/ reads the refresh_token cookie from the session's cookie + jar.""" @pytest.fixture(autouse=True) def _session_data(self): data = _obtain_tokens() @@ -160,16 +179,19 @@ def _session_data(self): self.session = data["session"] def test_refresh_returns_200(self): + """POST /api/token/refresh/ with the session's refresh_token cookie returns 200.""" r = self.session.post(f"{BASE}/api/token/refresh/", timeout=TIMEOUT) assert r.status_code == 200 def test_refresh_returns_new_access_token(self): + """The refresh response sets a new, non-empty access_token cookie.""" r = self.session.post(f"{BASE}/api/token/refresh/", timeout=TIMEOUT) assert r.status_code == 200 # New access_token is set in a cookie assert r.cookies.get("access_token"), "No new access_token cookie in refresh response" def test_refresh_with_invalid_token_returns_401(self): + """A refresh with an invalid refresh_token cookie returns 401.""" bad_session = requests.Session() bad_session.headers.update(_TRUSTED_PROXY_HEADERS) bad_session.cookies.set("refresh_token", "invalid.token.value") @@ -180,6 +202,8 @@ def test_refresh_with_invalid_token_returns_401(self): # ── Projects API ────────────────────────────────────────────────────────────── class TestLimsProjects: + """GET /api/projects/ requires authentication; the access token cookie value is sent + as a Bearer header.""" @pytest.fixture(autouse=True) def _auth_headers(self): data = _obtain_tokens() @@ -187,20 +211,24 @@ def _auth_headers(self): self.headers = {"Authorization": f"Bearer {data['access']}"} def test_projects_list_returns_200(self): + """An authenticated GET /api/projects/ returns 200.""" r = _get("/api/projects/", self.headers) assert r.status_code == 200 def test_projects_list_returns_json_array_or_paginated(self): + """The projects response is JSON that is either a list or a paginated object.""" r = _get("/api/projects/", self.headers) data = r.json() # DRF can return a plain list or a paginated {"count": ..., "results": [...]} assert isinstance(data, (list, dict)) def test_projects_list_without_auth_returns_401_or_403(self): + """GET /api/projects/ with no credentials returns 401 or 403.""" r = _get("/api/projects/") assert r.status_code in (401, 403) def test_projects_list_with_bad_token_returns_401(self): + """A bad Bearer token on /api/projects/ returns 401 or 403.""" r = _get("/api/projects/", {"Authorization": "Bearer bad-token"}) assert r.status_code in (401, 403) @@ -208,16 +236,19 @@ def test_projects_list_with_bad_token_returns_401(self): # ── Samples API ─────────────────────────────────────────────────────────────── class TestLimsSamples: + """GET /api/samples/ requires authentication.""" @pytest.fixture(autouse=True) def _auth_headers(self): data = _obtain_tokens() self.headers = {"Authorization": f"Bearer {data['access']}"} def test_samples_list_returns_200(self): + """An authenticated GET /api/samples/ returns 200.""" r = _get("/api/samples/", self.headers) assert r.status_code == 200 def test_samples_list_without_auth_returns_401(self): + """GET /api/samples/ with no credentials returns 401 or 403.""" r = _get("/api/samples/") assert r.status_code in (401, 403) @@ -225,21 +256,26 @@ def test_samples_list_without_auth_returns_401(self): # ── Current user (me) ───────────────────────────────────────────────────────── class TestLimsMe: + """GET /api/auth/me/ returns the current user when authenticated and is denied + otherwise.""" @pytest.fixture(autouse=True) def _auth_headers(self): data = _obtain_tokens() self.headers = {"Authorization": f"Bearer {data['access']}"} def test_me_returns_200(self): + """An authenticated GET /api/auth/me/ returns 200.""" r = _get("/api/auth/me/", self.headers) assert r.status_code == 200 def test_me_returns_username_field(self): + """The /api/auth/me/ response includes a username or an email field.""" r = _get("/api/auth/me/", self.headers) data = r.json() assert "username" in data or "email" in data def test_me_without_auth_returns_401(self): + """GET /api/auth/me/ with no credentials returns 401 or 403.""" r = _get("/api/auth/me/") assert r.status_code in (401, 403) @@ -247,16 +283,19 @@ def test_me_without_auth_returns_401(self): # ── Stats ───────────────────────────────────────────────────────────────────── class TestLimsStats: + """GET /api/stats/ requires authentication.""" @pytest.fixture(autouse=True) def _auth_headers(self): data = _obtain_tokens() self.headers = {"Authorization": f"Bearer {data['access']}"} def test_stats_returns_200(self): + """An authenticated GET /api/stats/ returns 200.""" r = _get("/api/stats/", self.headers) assert r.status_code == 200 def test_stats_without_auth_returns_401(self): + """GET /api/stats/ with no credentials returns 401 or 403.""" r = _get("/api/stats/") assert r.status_code in (401, 403) @@ -264,7 +303,12 @@ def test_stats_without_auth_returns_401(self): # ── Full round-trip ─────────────────────────────────────────────────────────── class TestLimsFullFlow: + """Login, list projects, refresh the access token and list projects again with the + refreshed token.""" def test_login_list_projects_refresh_logout(self): + """With the cached login, list projects with the access token, refresh through + the cookie session to obtain a new access_token, then list projects again with + it (each step 200). No logout step is exercised despite the name.""" # Use cached login to avoid rate-limiting — we already verified login works above. data = _obtain_tokens() access = data["access"] diff --git a/tests/integration/test_org_switching.py b/tests/integration/test_org_switching.py index 8fcad444..5bd41b09 100644 --- a/tests/integration/test_org_switching.py +++ b/tests/integration/test_org_switching.py @@ -18,6 +18,9 @@ whatever org they create (app/services/org_service.py), so two POST /orgs calls by the same user are enough to set up multi-org membership -- no /orgs/{id}/invite round trip needed. + +Developer: + Manish Kumar """ import uuid @@ -30,6 +33,8 @@ def _register_and_login() -> dict: + """Registers a throwaway user with a unique email against the auth service and logs + in, returning the token response.""" email = f"itest-orgswitch-{uuid.uuid4().hex}@example.com" password = "S3curePass!1" @@ -46,6 +51,8 @@ def _headers(access_token: str) -> dict: def _create_org(access_token: str) -> dict: + """Creates an organization with a unique slug as the given user (asserting HTTP 201) + and returns its JSON; the creator becomes an org admin member.""" slug = f"itest-org-{uuid.uuid4().hex[:12]}" r = requests.post( f"{BASE}/orgs", @@ -58,7 +65,10 @@ def _create_org(access_token: str) -> dict: class TestMultiOrgUserCanActOnEachOwnOrg: + """A user who created several orgs can act on each with the same access token, + because authorization is re-derived per request from the org id in the URL path.""" def test_user_can_get_first_org_they_created(self): + """GET /orgs/{id} for an org the user created returns 200 with that org's id.""" tokens = _register_and_login() org_a = _create_org(tokens["access_token"]) @@ -67,6 +77,8 @@ def test_user_can_get_first_org_they_created(self): assert r.json()["id"] == org_a["id"] def test_same_token_can_act_on_a_second_org_too(self): + """The same access token can GET two different orgs the user created, each + returning 200 with distinct ids.""" tokens = _register_and_login() org_a = _create_org(tokens["access_token"]) org_b = _create_org(tokens["access_token"]) @@ -79,6 +91,7 @@ def test_same_token_can_act_on_a_second_org_too(self): assert r_a.json()["id"] != r_b.json()["id"] def test_list_my_orgs_includes_both_created_orgs(self): + """GET /orgs lists both orgs the user created.""" tokens = _register_and_login() org_a = _create_org(tokens["access_token"]) org_b = _create_org(tokens["access_token"]) @@ -91,7 +104,9 @@ def test_list_my_orgs_includes_both_created_orgs(self): class TestNonMemberCannotAccessAnotherUsersOrg: + """A user who is not a member of an org cannot see it.""" def test_get_org_owned_by_a_different_user_returns_404(self): + """GET /orgs/{id} for another user's org returns 404 for a non-member.""" owner_tokens = _register_and_login() outsider_tokens = _register_and_login() @@ -105,6 +120,7 @@ def test_get_org_owned_by_a_different_user_returns_404(self): assert r.status_code == 404 def test_outsiders_org_list_does_not_include_owners_org(self): + """A non-member's GET /orgs succeeds but does not include the other user's org.""" owner_tokens = _register_and_login() outsider_tokens = _register_and_login() diff --git a/tests/integration/test_rag_integration.py b/tests/integration/test_rag_integration.py index 6d55e7a4..e20ce1f0 100644 --- a/tests/integration/test_rag_integration.py +++ b/tests/integration/test_rag_integration.py @@ -16,6 +16,9 @@ against the central auth service and reusing its access_token. Run: pytest tests/integration/test_rag_integration.py -v + +Developer: + Manish Kumar """ import uuid @@ -31,6 +34,8 @@ # ── Helpers ─────────────────────────────────────────────────────────────────── def _get(path: str, auth: bool = True) -> requests.Response: + """GETs a RAG path, optionally with the static RAGBIO_API_KEY bearer token, and + retries once on a timeout.""" headers = {"Authorization": f"Bearer {RAGBIO_API_KEY}"} if auth else {} # /v1/studies in particular sees occasional multi-second latency spikes # in this environment; one retry absorbs those without masking a real @@ -68,6 +73,8 @@ def _query_jwt() -> str: def _post_query(body: dict) -> requests.Response: + """POSTs to /v1/query with a freshly minted IAM JWT; each call registers a new + throwaway user in the auth service.""" headers = {"Authorization": f"Bearer {_query_jwt()}"} return requests.post(f"{BASE}/v1/query", json=body, headers=headers, timeout=TIMEOUT) @@ -75,19 +82,25 @@ def _post_query(body: dict) -> requests.Response: # ── Health ──────────────────────────────────────────────────────────────────── class TestRagHealth: + """RAG /health probe: responds 200 with status ok and a version, and needs no + credentials.""" def test_health_returns_200(self): + """GET /health on the RAG service returns HTTP 200 without credentials.""" r = _get("/health", auth=False) assert r.status_code == 200 def test_health_status_ok(self): + """The /health body reports status ok.""" r = _get("/health", auth=False) assert r.json()["status"] == "ok" def test_health_has_version(self): + """The /health body includes a version field.""" r = _get("/health", auth=False) assert "version" in r.json() def test_health_does_not_require_auth(self): + """GET /health with no Authorization header returns 200.""" # Health must be reachable without any credentials r = requests.get(f"{BASE}/health", timeout=TIMEOUT) assert r.status_code == 200 @@ -96,7 +109,9 @@ def test_health_does_not_require_auth(self): # ── Authentication ──────────────────────────────────────────────────────────── class TestRagAuth: + """RAG endpoints reject requests that carry no credentials.""" def test_ingest_without_auth_returns_401_or_403(self): + """POST /v1/ingest with no credentials returns 401 or 403.""" r = _post( "/v1/ingest", {"study": "test", "search_query": "BRCA1"}, @@ -105,6 +120,7 @@ def test_ingest_without_auth_returns_401_or_403(self): assert r.status_code in (401, 403) def test_query_without_auth_returns_401_or_403(self): + """POST /v1/query with no credentials returns 401 or 403.""" r = _post( "/v1/query", {"query": "BRCA1", "study": "default", "top_k": 3, "mode": "rag"}, @@ -116,22 +132,30 @@ def test_query_without_auth_returns_401_or_403(self): # ── Studies ─────────────────────────────────────────────────────────────────── class TestRagStudies: + """GET /v1/studies response shape; the slow response is fetched once per class and + shared.""" # /v1/studies is slow (~7-9s); fetch it once per class instead of once # per test to keep the suite fast and avoid piling up timeout risk. @pytest.fixture(scope="class") def studies_response(self): + """Class-scoped fetch of /v1/studies, requested once because the endpoint is + slow.""" return _get("/v1/studies") def test_studies_returns_200(self, studies_response): + """GET /v1/studies with the static API key returns HTTP 200.""" assert studies_response.status_code == 200 def test_studies_returns_dict_with_studies_key(self, studies_response): + """The /v1/studies response body has a studies key.""" assert "studies" in studies_response.json() def test_studies_is_list(self, studies_response): + """The studies value in the /v1/studies response is a list.""" assert isinstance(studies_response.json()["studies"], list) def test_each_study_has_name_field(self, studies_response): + """Every entry in the studies list has a name field.""" for study in studies_response.json()["studies"]: assert "name" in study @@ -139,21 +163,31 @@ def test_each_study_has_name_field(self, studies_response): # ── Cache ───────────────────────────────────────────────────────────────────── class TestRagCache: + """GET /v1/cache response shape.""" @pytest.fixture(scope="class") def cache_response(self): + """Class-scoped fetch of /v1/cache shared by the cache tests.""" return _get("/v1/cache") def test_cache_returns_200(self, cache_response): + """GET /v1/cache with the static API key returns HTTP 200.""" assert cache_response.status_code == 200 def test_cache_returns_dict_with_cache_key(self, cache_response): + """The /v1/cache response body has a cache key.""" assert "cache" in cache_response.json() # ── Query ───────────────────────────────────────────────────────────────────── class TestRagQuery: + """POST /v1/query with an IAM-issued JWT: accepted modes and study selectors, + response shape when the status is 200, and 422 validation of bad input. Statuses 404 + and 500 are tolerated where the environment may lack indexed studies or a configured + embedding model.""" def test_query_pmids_only_mode_returns_200(self): + """A pmids_only query returns 200, 404 or 500; 500 is tolerated when the + embedding model is not configured.""" r = _post_query({ "query": "BRCA1 cancer", "study": "*", @@ -165,6 +199,8 @@ def test_query_pmids_only_mode_returns_200(self): assert r.status_code in (200, 404, 500) def test_query_pmids_only_returns_mode_field(self): + """When a pmids_only query returns 200, the body reports mode pmids_only; other + statuses are not checked.""" r = _post_query({ "query": "BRCA1 cancer", "study": "*", @@ -176,6 +212,8 @@ def test_query_pmids_only_returns_mode_field(self): assert r.json()["mode"] == "pmids_only" def test_query_pmids_only_returns_pmids_list(self): + """When a pmids_only query returns 200, the body has a pmids list; other + statuses are not checked.""" r = _post_query({ "query": "BRCA1", "study": "*", @@ -187,6 +225,8 @@ def test_query_pmids_only_returns_pmids_list(self): assert isinstance(r.json()["pmids"], list) def test_query_rag_mode_returns_events_key(self): + """When a rag-mode query returns 200, the body has an events key; other statuses + are not checked.""" r = _post_query({ "query": "BRCA1", "study": "*", @@ -197,6 +237,8 @@ def test_query_rag_mode_returns_events_key(self): assert "events" in r.json() def test_query_no_study_data_returns_404_or_200(self): + """A rag-mode query over the '*' study returns 200, 404 (no studies indexed) or + 500 (embedding model not configured).""" # When no studies are indexed, 404 is acceptable; 500 when Ollama not configured r = _post_query({ "query": "BRCA1 cancer drug therapy", @@ -207,6 +249,7 @@ def test_query_no_study_data_returns_404_or_200(self): assert r.status_code in (200, 404, 500) def test_query_invalid_mode_returns_422(self): + """A query with an unsupported mode value returns 422.""" r = _post_query({ "query": "BRCA1", "study": "default", @@ -216,6 +259,7 @@ def test_query_invalid_mode_returns_422(self): assert r.status_code == 422 def test_query_top_k_zero_returns_422(self): + """A query with top_k=0 returns 422.""" r = _post_query({ "query": "BRCA1", "study": "default", @@ -225,10 +269,13 @@ def test_query_top_k_zero_returns_422(self): assert r.status_code == 422 def test_query_missing_required_fields_returns_422(self): + """A query body missing the required query field returns 422.""" r = _post_query({"study": "default"}) assert r.status_code == 422 def test_query_wildcard_study_accepted(self): + """The '*' study selector is accepted: a pmids_only query returns 200, 404 or + 500, not a validation error.""" r = _post_query({ "query": "cancer", "study": "*", @@ -238,6 +285,8 @@ def test_query_wildcard_study_accepted(self): assert r.status_code in (200, 404, 500) def test_query_all_study_accepted(self): + """The 'all' study selector is accepted: a pmids_only query returns 200, 404 or + 500, not a validation error.""" r = _post_query({ "query": "cancer", "study": "all", @@ -250,7 +299,10 @@ def test_query_all_study_accepted(self): # ── Full round-trip ─────────────────────────────────────────────────────────── class TestRagFullFlow: + """Health check, then study listing, then a query, in one sequence.""" def test_health_then_studies_then_query(self): + """Check /health, list studies, then query the first indexed study, or a + nonexistent one when none are indexed; the query may return 200, 404 or 500.""" # Health check r1 = _get("/health", auth=False) assert r1.status_code == 200 diff --git a/tests/integration/test_service_to_service_oauth.py b/tests/integration/test_service_to_service_oauth.py index bc960b10..7a938054 100644 --- a/tests/integration/test_service_to_service_oauth.py +++ b/tests/integration/test_service_to_service_oauth.py @@ -20,6 +20,9 @@ the returned JWT WITHOUT verifying its signature (no shared secret is exposed to a black-box test) purely to assert the claims a real consumer would need are actually present. + +Developer: + Manish Kumar """ import uuid @@ -33,6 +36,8 @@ def _register_and_login() -> dict: + """Registers a throwaway user with a unique email against the auth service and logs + in, returning the token response.""" email = f"itest-svc2svc-{uuid.uuid4().hex}@example.com" password = "S3curePass!1" @@ -77,6 +82,8 @@ def _setup_client(scopes: list) -> dict: def _mint_token(client_id: str, client_secret: str, scope: str | None = None) -> requests.Response: + """POSTs a form-encoded client_credentials grant to /oauth/token, optionally + requesting a scope, and returns the response.""" data = {"grant_type": "client_credentials", "client_id": client_id, "client_secret": client_secret} if scope is not None: data["scope"] = scope @@ -84,12 +91,17 @@ def _mint_token(client_id: str, client_secret: str, scope: str | None = None) -> class TestClientCredentialsHappyPath: + """Successful client_credentials token issue: status, token shape, scope defaulting + and subsetting, and claims that identify a service rather than a user.""" def test_mint_token_returns_200(self): + """A registered client's client_credentials request to /oauth/token returns 200.""" client = _setup_client(["read:reports", "write:reports"]) r = _mint_token(client["client_id"], client["client_secret"]) assert r.status_code == 200 def test_mint_token_returns_access_token_and_bearer_type(self): + """The token response has a non-empty access_token, token_type bearer and a + positive expires_in.""" client = _setup_client(["read:reports"]) r = _mint_token(client["client_id"], client["client_secret"]) body = r.json() @@ -98,17 +110,24 @@ def test_mint_token_returns_access_token_and_bearer_type(self): assert body["expires_in"] > 0 def test_token_defaults_to_all_granted_scopes_when_scope_omitted(self): + """With no scope requested, the issued token's scope lists every scope granted + to the client.""" client = _setup_client(["read:reports", "write:reports"]) r = _mint_token(client["client_id"], client["client_secret"]) assert set(r.json()["scope"].split()) == {"read:reports", "write:reports"} def test_token_can_request_a_subset_of_granted_scopes(self): + """Requesting a subset of the granted scopes returns 200 with exactly that + scope.""" client = _setup_client(["read:reports", "write:reports"]) r = _mint_token(client["client_id"], client["client_secret"], scope="read:reports") assert r.status_code == 200 assert r.json()["scope"] == "read:reports" def test_minted_token_claims_identify_a_service_not_a_user(self): + """The decoded token (signature not verified) has auth_method + client_credentials, the client_id and the granted scopes, and carries no sub or + email claim.""" client = _setup_client(["read:reports"]) r = _mint_token(client["client_id"], client["client_secret"]) payload = pyjwt.decode(r.json()["access_token"], options={"verify_signature": False}) @@ -121,21 +140,28 @@ def test_minted_token_claims_identify_a_service_not_a_user(self): class TestClientCredentialsRejections: + """Rejection paths of /oauth/token: bad credentials return 401, while ungranted + scopes, unsupported grant types and missing credentials return 400.""" def test_wrong_client_secret_returns_401(self): + """A wrong client_secret for a real client returns 401.""" client = _setup_client(["read:reports"]) r = _mint_token(client["client_id"], "not-the-real-secret") assert r.status_code == 401 def test_unknown_client_id_returns_401(self): + """An unknown client_id returns 401.""" r = _mint_token(f"nonexistent-{uuid.uuid4().hex}", "whatever") assert r.status_code == 401 def test_requesting_ungranted_scope_returns_400(self): + """Requesting a scope the client was not granted returns 400.""" client = _setup_client(["read:reports"]) r = _mint_token(client["client_id"], client["client_secret"], scope="delete:everything") assert r.status_code == 400 def test_unsupported_grant_type_returns_400(self): + """The password grant type is rejected with 400 even with valid client + credentials.""" client = _setup_client(["read:reports"]) r = requests.post( f"{BASE}/oauth/token", @@ -149,12 +175,17 @@ def test_unsupported_grant_type_returns_400(self): assert r.status_code == 400 def test_missing_credentials_returns_400(self): + """A client_credentials request with no client credentials returns 400.""" r = requests.post(f"{BASE}/oauth/token", data={"grant_type": "client_credentials"}, timeout=TIMEOUT) assert r.status_code == 400 class TestClientCredentialsViaHttpBasic: + """Client credentials supplied through HTTP Basic authentication instead of form + fields.""" def test_basic_auth_credentials_are_accepted(self): + """HTTP Basic client credentials are accepted: the token request returns 200 + with an access_token.""" client = _setup_client(["read:reports"]) r = requests.post( f"{BASE}/oauth/token", diff --git a/tests/integration/test_services_health.py b/tests/integration/test_services_health.py index cfa06c42..1bf9a817 100644 --- a/tests/integration/test_services_health.py +++ b/tests/integration/test_services_health.py @@ -6,6 +6,9 @@ • Directly on their native ports for authoritative health checks Run: pytest tests/integration/test_services_health.py -v + +Developer: + Manish Kumar """ import uuid @@ -44,6 +47,8 @@ def _bearer_token() -> str: def _is_up(url: str) -> bool: + """Returns True when the URL answers with a status below 500; a connection error or + timeout counts as down.""" try: r = _get(url) return r.status_code < 500 @@ -54,15 +59,20 @@ def _is_up(url: str) -> bool: # ── Nginx router ────────────────────────────────────────────────────────────── class TestNginxRouter: + """The nginx router's /_health endpoint responds 200 with status ok and a router + field.""" def test_nginx_health_returns_200(self): + """GET /_health on the nginx router returns HTTP 200.""" r = _get(f"{BASE_URL}/_health") assert r.status_code == 200 def test_nginx_health_body_has_status_ok(self): + """The router /_health body reports status ok.""" r = _get(f"{BASE_URL}/_health") assert r.json()["status"] == "ok" def test_nginx_health_body_has_router_field(self): + """The router /_health body includes a router field.""" r = _get(f"{BASE_URL}/_health") assert "router" in r.json() @@ -70,15 +80,20 @@ def test_nginx_health_body_has_router_field(self): # ── Auth service ────────────────────────────────────────────────────────────── class TestAuthServiceHealth: + """Auth service health, directly on port 8001 and through the router at + /_svc/auth/health.""" def test_direct_health_returns_200(self): + """GET /health directly on the auth service returns 200.""" r = requests.get("http://localhost:8001/health", timeout=TIMEOUT) assert r.status_code == 200 def test_direct_health_body(self): + """The direct auth /health body reports status ok.""" r = requests.get("http://localhost:8001/health", timeout=TIMEOUT) assert r.json().get("status") == "ok" def test_via_nginx_health(self): + """GET /_svc/auth/health through the router returns 200.""" r = _get(f"{BASE_URL}/_svc/auth/health") assert r.status_code == 200 @@ -86,11 +101,15 @@ def test_via_nginx_health(self): # ── Workbench (Django) ──────────────────────────────────────────────────────── class TestWorkbenchHealth: + """The Workbench root path is reachable, meaning a status below 500, directly on + port 8000 and through the router.""" def test_direct_root_reachable(self): + """GET http://localhost:8000/ returns a status below 500.""" r = requests.get("http://localhost:8000/", timeout=TIMEOUT) assert r.status_code < 500 def test_via_nginx_root_reachable(self): + """GET / through the router returns a status below 500.""" r = _get(f"{BASE_URL}/") assert r.status_code < 500 @@ -98,15 +117,20 @@ def test_via_nginx_root_reachable(self): # ── LIMS ────────────────────────────────────────────────────────────────────── class TestLimsHealth: + """LIMS /healthz and /readyz respond 200 directly on port 7000, and /healthz + responds 200 through the router at /_svc/lims/.""" def test_direct_healthz_returns_200(self): + """GET /healthz directly on LIMS returns 200.""" r = requests.get("http://localhost:7000/healthz", timeout=TIMEOUT) assert r.status_code == 200 def test_direct_readyz_returns_200(self): + """GET /readyz directly on LIMS returns 200.""" r = requests.get("http://localhost:7000/readyz", timeout=TIMEOUT) assert r.status_code == 200 def test_via_nginx_healthz(self): + """GET /_svc/lims/healthz through the router returns 200.""" r = requests.get(f"{BASE_URL}/_svc/lims/healthz", timeout=TIMEOUT) assert r.status_code == 200 @@ -114,15 +138,20 @@ def test_via_nginx_healthz(self): # ── TES ─────────────────────────────────────────────────────────────────────── class TestTesHealth: + """TES health directly on port 8081: /health returns 200 with ok=true, and the tools + list endpoint returns 200.""" def test_direct_health_returns_200(self): + """GET /health directly on TES returns 200.""" r = requests.get("http://localhost:8081/health", timeout=TIMEOUT) assert r.status_code == 200 def test_direct_health_body_ok_true(self): + """The direct TES /health body has ok set to true.""" r = requests.get("http://localhost:8081/health", timeout=TIMEOUT) assert r.json().get("ok") is True def test_direct_tools_list_reachable(self): + """GET /api/tools directly on TES returns 200.""" r = requests.get("http://localhost:8081/api/tools", timeout=TIMEOUT) assert r.status_code == 200 @@ -130,16 +159,20 @@ def test_direct_tools_list_reachable(self): # ── RAG ─────────────────────────────────────────────────────────────────────── class TestRagHealth: + """RAG /health directly on port 8090 and through the router at /_svc/rag/health.""" def test_direct_health_returns_200(self): + """GET /health directly on RAG returns 200.""" r = requests.get("http://localhost:8090/health", timeout=TIMEOUT) assert r.status_code == 200 def test_direct_health_body(self): + """The direct RAG /health body reports status ok.""" r = requests.get("http://localhost:8090/health", timeout=TIMEOUT) data = r.json() assert data.get("status") == "ok" def test_via_nginx_health(self): + """GET /_svc/rag/health through the router returns 200.""" r = _get(f"{BASE_URL}/_svc/rag/health") assert r.status_code == 200 @@ -147,11 +180,15 @@ def test_via_nginx_health(self): # ── API Gateway ─────────────────────────────────────────────────────────────── class TestApiGatewayHealth: + """API Gateway /health directly on port 8080 and through the router at + /_svc/gateway/health.""" def test_direct_health_returns_200(self): + """GET /health directly on the API Gateway returns 200.""" r = requests.get("http://localhost:8080/health", timeout=TIMEOUT) assert r.status_code == 200 def test_via_nginx_health(self): + """GET /_svc/gateway/health through the router returns 200.""" r = _get(f"{BASE_URL}/_svc/gateway/health") assert r.status_code == 200 @@ -160,11 +197,15 @@ def test_via_nginx_health(self): # The policy engine exposes no /health route; /openapi.json is only a liveness check. class TestPolicyEngineHealth: + """Policy engine liveness through /openapi.json, since it exposes no /health route: + directly on port 8002 and through /_svc/policy/.""" def test_direct_openapi_returns_200(self): + """GET /openapi.json directly on the policy engine returns 200.""" r = requests.get("http://localhost:8002/openapi.json", timeout=TIMEOUT) assert r.status_code == 200 def test_via_nginx_health(self): + """GET /_svc/policy/openapi.json through the router returns 200.""" r = _get(f"{BASE_URL}/_svc/policy/openapi.json") assert r.status_code == 200 @@ -173,11 +214,15 @@ def test_via_nginx_health(self): # Same as policy engine: no /health route, use /openapi.json. class TestHpcPolicyEngineHealth: + """HPC policy engine liveness through /openapi.json, since it exposes no /health + route: directly on port 8003 and through /_svc/hpc/.""" def test_direct_openapi_returns_200(self): + """GET /openapi.json directly on the HPC policy engine returns 200.""" r = requests.get("http://localhost:8003/openapi.json", timeout=TIMEOUT) assert r.status_code == 200 def test_via_nginx_health(self): + """GET /_svc/hpc/openapi.json through the router returns 200.""" r = _get(f"{BASE_URL}/_svc/hpc/openapi.json") assert r.status_code == 200 @@ -185,11 +230,15 @@ def test_via_nginx_health(self): # ── Security Audit ──────────────────────────────────────────────────────────── class TestSecurityAuditHealth: + """Security-audit /health directly on port 8004 and through the router at + /_svc/audit/health.""" def test_direct_health_returns_200(self): + """GET /health directly on the security-audit service returns 200.""" r = requests.get("http://localhost:8004/health", timeout=TIMEOUT) assert r.status_code == 200 def test_via_nginx_health(self): + """GET /_svc/audit/health through the router returns 200.""" r = _get(f"{BASE_URL}/_svc/audit/health") assert r.status_code == 200 @@ -197,11 +246,15 @@ def test_via_nginx_health(self): # ── Control Center ──────────────────────────────────────────────────────────── class TestControlCenterHealth: + """Control Center /health is reachable, meaning a status below 500, directly on port + 7070 and through /_svc/control/health.""" def test_direct_reachable(self): + """GET /health directly on Control Center returns a status below 500.""" r = requests.get("http://localhost:7070/health", timeout=TIMEOUT) assert r.status_code < 500 def test_via_nginx_reachable(self): + """GET /_svc/control/health through the router returns a status below 500.""" r = _get(f"{BASE_URL}/_svc/control/health") assert r.status_code < 500 @@ -209,11 +262,17 @@ def test_via_nginx_reachable(self): # ── Toolserver ──────────────────────────────────────────────────────────────── class TestToolserverHealth: + """Toolserver /health returns 200 directly on port 9090; through the router it is + auth_request-gated end to end and needs a real bearer token.""" def test_direct_health_returns_200(self): + """GET /health directly on the toolserver returns 200.""" r = requests.get("http://localhost:9090/health", timeout=TIMEOUT) assert r.status_code == 200 def test_via_nginx_health(self): + """GET /_svc/toolserver/health through the router with a bearer token from the + central auth service returns 200; a throwaway user is registered to obtain the + token.""" # /_svc/toolserver is auth_request-gated end to end (including this # health path -- unlike Control Center's own health/summary/services # paths just above, toolserver has no unauthenticated carve-out) since @@ -229,11 +288,15 @@ def test_via_nginx_health(self): # ── Model Registry ──────────────────────────────────────────────────────────── class TestModelRegistryHealth: + """Model Registry /health is reachable, meaning a status below 500, directly on port + 8095 and through /_svc/modelregistry/health.""" def test_direct_reachable(self): + """GET /health directly on the model registry returns a status below 500.""" r = requests.get("http://localhost:8095/health", timeout=TIMEOUT) assert r.status_code < 500 def test_via_nginx_reachable(self): + """GET /_svc/modelregistry/health through the router returns a status below 500.""" r = _get(f"{BASE_URL}/_svc/modelregistry/health") assert r.status_code < 500 @@ -241,7 +304,9 @@ def test_via_nginx_reachable(self): # ── Workflow Bundles ────────────────────────────────────────────────────────── class TestWorkflowBundlesHealth: + """Workflow Bundles /health on port 8098 is reachable, meaning a status below 500.""" def test_direct_reachable(self): + """GET /health directly on workflow-bundles returns a status below 500.""" r = requests.get("http://localhost:8098/health", timeout=TIMEOUT) assert r.status_code < 500 @@ -250,11 +315,15 @@ def test_direct_reachable(self): # OPA is part of the primary 40-service Compose profile and is required here. class TestOpaHealth: + """OPA /health returns 200 directly on port 8181 and through the router at + /_svc/opa/health.""" def test_direct_health_returns_200(self): + """GET /health directly on OPA returns 200.""" r = requests.get("http://localhost:8181/health", timeout=TIMEOUT) assert r.status_code == 200 def test_via_nginx_health(self): + """GET /_svc/opa/health through the router returns 200.""" r = _get(f"{BASE_URL}/_svc/opa/health") assert r.status_code == 200 @@ -262,11 +331,15 @@ def test_via_nginx_health(self): # ── Grafana ─────────────────────────────────────────────────────────────────── class TestGrafanaHealth: + """Grafana /api/health returns 200 directly on port 3000 and through the router at + /_svc/monitor/api/health.""" def test_direct_reachable(self): + """GET /api/health directly on Grafana returns 200.""" r = requests.get("http://localhost:3000/api/health", timeout=TIMEOUT) assert r.status_code == 200 def test_via_nginx_reachable(self): + """GET /_svc/monitor/api/health through the router returns 200.""" r = _get(f"{BASE_URL}/_svc/monitor/api/health") assert r.status_code == 200 @@ -291,10 +364,14 @@ def test_via_nginx_reachable(self): @pytest.mark.parametrize("name,url", SERVICES) def test_service_is_up(name, url): + """Parametrized over the SERVICES list: each service's health or openapi URL is + reachable, meaning the request completes with a status below 500.""" assert _is_up(url), f"Service '{name}' is not reachable at {url}" def test_is_up_returns_false_for_unreachable_url(): + """_is_up returns False, rather than raising, for a URL nothing listens on + (localhost port 1).""" # Port 1 is reserved and nothing listens there, so this should raise a # RequestException that _is_up() swallows and turns into False. assert _is_up("http://localhost:1/health") is False diff --git a/tests/integration/test_tes_integration.py b/tests/integration/test_tes_integration.py index d79e7e5e..9aab87ba 100644 --- a/tests/integration/test_tes_integration.py +++ b/tests/integration/test_tes_integration.py @@ -15,6 +15,9 @@ the run reaches COMPLETE or FAILED. Run: pytest tests/integration/test_tes_integration.py -v + +Developer: + Manish Kumar """ import os @@ -66,11 +69,14 @@ def _poll_run(run_id: str) -> dict: # ── Health ──────────────────────────────────────────────────────────────────── class TestTesHealth: + """TES /health responds 200 with ok set to true.""" def test_health_returns_200(self): + """GET /health on TES returns HTTP 200.""" r = _get("/health") assert r.status_code == 200 def test_health_body_ok_true(self): + """The TES /health body has ok set to true.""" r = _get("/health") assert r.json()["ok"] is True @@ -78,19 +84,25 @@ def test_health_body_ok_true(self): # ── Tools registry ──────────────────────────────────────────────────────────── class TestTesTools: + """The TES tool registry at /api/tools lists tools, including the built-in echo_test + tool.""" def test_list_tools_returns_200(self): + """GET /api/tools returns HTTP 200.""" r = _get("/api/tools") assert r.status_code == 200 def test_list_tools_returns_list(self): + """The /api/tools response body is a list.""" r = _get("/api/tools") assert isinstance(r.json(), list) def test_list_tools_not_empty(self): + """The /api/tools list contains at least one tool.""" r = _get("/api/tools") assert len(r.json()) > 0 def test_echo_test_tool_is_registered(self): + """echo_test is among the tool ids registered with TES.""" r = _get("/api/tools") tool_ids = {t.get("tool_id") or t.get("id", "") for t in r.json()} assert ECHO_TOOL_ID in tool_ids, ( @@ -98,13 +110,16 @@ def test_echo_test_tool_is_registered(self): ) def test_each_tool_has_tool_id_field(self): + """Every registered tool entry has a tool_id or id field.""" r = _get("/api/tools") for tool in r.json(): assert "tool_id" in tool or "id" in tool class TestTesAuthentication: + """TES run submission is protected: a request without a token is rejected.""" def test_submit_without_token_is_protected(self): + """POST /api/runs/submit with no bearer token returns 401 or 403.""" r = requests.post( f"{BASE}/api/runs/submit", json={"tool_id": ECHO_TOOL_ID, "inputs": {"message": ECHO_TEXT}, "resources": {}}, @@ -117,11 +132,15 @@ def test_submit_without_token_is_protected(self): @pytest.mark.skipif(not TES_TOKEN, reason="OMNIBIOAI_TES_TOKEN is not configured") class TestTesRunsList: + """GET /api/runs with the OMNIBIOAI_TES_TOKEN bearer token; skipped when the token + is not configured.""" def test_list_runs_returns_200(self): + """An authenticated GET /api/runs returns HTTP 200.""" r = _get("/api/runs") assert r.status_code == 200 def test_list_runs_returns_list(self): + """The /api/runs response body is a list or an object.""" r = _get("/api/runs") data = r.json() assert isinstance(data, (list, dict)) @@ -131,7 +150,11 @@ def test_list_runs_returns_list(self): @pytest.mark.skipif(not TES_TOKEN, reason="OMNIBIOAI_TES_TOKEN is not configured") class TestTesSubmit: + """POST /api/runs/submit with the TES token: a valid echo_test submission, and + rejection of an unknown tool or a missing tool_id. Successful submissions create + runs on the live TES service. Skipped when the token is not configured.""" def test_submit_echo_test_returns_200(self): + """Submitting an echo_test run returns HTTP 200.""" r = _post( "/api/runs/submit", {"tool_id": ECHO_TOOL_ID, "inputs": {"message": ECHO_TEXT}, "resources": {}}, @@ -139,6 +162,7 @@ def test_submit_echo_test_returns_200(self): assert r.status_code == 200 def test_submit_echo_test_returns_run_id(self): + """The submit response includes a non-empty run_id.""" r = _post( "/api/runs/submit", {"tool_id": ECHO_TOOL_ID, "inputs": {"message": ECHO_TEXT}, "resources": {}}, @@ -148,6 +172,7 @@ def test_submit_echo_test_returns_run_id(self): assert data["run_id"] def test_submit_unknown_tool_returns_error(self): + """Submitting an unknown tool_id returns 400, 404 or 422.""" r = _post( "/api/runs/submit", {"tool_id": "nonexistent_tool_xyz", "inputs": {}, "resources": {}}, @@ -155,6 +180,7 @@ def test_submit_unknown_tool_returns_error(self): assert r.status_code in (400, 404, 422) def test_submit_missing_tool_id_returns_422(self): + """Submitting a body with no tool_id returns 422.""" r = _post("/api/runs/submit", {"inputs": {}, "resources": {}}) assert r.status_code == 422 @@ -163,8 +189,12 @@ def test_submit_missing_tool_id_returns_422(self): @pytest.mark.skipif(not TES_TOKEN, reason="OMNIBIOAI_TES_TOKEN is not configured") class TestTesRunStatus: + """GET /api/runs/{run_id} for a submitted run and for a nonexistent run. Skipped + when the TES token is not configured.""" @pytest.fixture(scope="class") def submitted_run(self): + """Class-scoped fixture: submits one echo_test run with the TES token and + returns its JSON, raising on a non-2xx response.""" r = requests.post( f"{BASE}/api/runs/submit", json={"tool_id": ECHO_TOOL_ID, "inputs": {"message": ECHO_TEXT}, "resources": {}}, @@ -175,16 +205,19 @@ def submitted_run(self): return r.json() def test_get_run_by_id_returns_200(self, submitted_run): + """GET /api/runs/{run_id} for a just-submitted run returns HTTP 200.""" run_id = submitted_run["run_id"] r = _get(f"/api/runs/{run_id}") assert r.status_code == 200 def test_get_run_contains_state_field(self, submitted_run): + """The run detail response includes a state field.""" run_id = submitted_run["run_id"] r = _get(f"/api/runs/{run_id}") assert "state" in r.json() def test_get_nonexistent_run_returns_404(self): + """GET /api/runs/{run_id} for a nonexistent run id returns 404.""" r = _get("/api/runs/run-does-not-exist-xyz") assert r.status_code == 404 @@ -193,7 +226,11 @@ def test_get_nonexistent_run_returns_404(self): @pytest.mark.skipif(not TES_TOKEN, reason="OMNIBIOAI_TES_TOKEN is not configured") class TestTesEndToEnd: + """Submit echo_test runs and poll them to a terminal state, then fetch logs and + results. Skipped when the TES token is not configured.""" def test_echo_test_completes_successfully(self): + """A submitted echo_test run reaches the state COMPLETED within the 60-second + polling limit.""" r = requests.post( f"{BASE}/api/runs/submit", json={"tool_id": ECHO_TOOL_ID, "inputs": {"message": ECHO_TEXT}, "resources": {}}, @@ -209,6 +246,8 @@ def test_echo_test_completes_successfully(self): ) def test_completed_run_logs_available(self): + """Once an echo_test run reaches a terminal state, GET /api/runs/{run_id}/logs + returns 200 or 204.""" r = requests.post( f"{BASE}/api/runs/submit", json={"tool_id": ECHO_TOOL_ID, "inputs": {"message": "log check"}, "resources": {}}, @@ -224,6 +263,8 @@ def test_completed_run_logs_available(self): assert r2.status_code in (200, 204) def test_completed_run_results_available(self): + """Once an echo_test run reaches a terminal state, GET + /api/runs/{run_id}/results returns 200 or 204.""" r = requests.post( f"{BASE}/api/runs/submit", json={"tool_id": ECHO_TOOL_ID, "inputs": {"message": "result check"}, "resources": {}}, diff --git a/tests/test_backup_health_check.py b/tests/test_backup_health_check.py index a9142fb8..7e6a691e 100644 --- a/tests/test_backup_health_check.py +++ b/tests/test_backup_health_check.py @@ -6,6 +6,9 @@ itself had a bug, so it must not share test fixtures/assumptions with it. See tests/test_backup_mysql.py and omnibioai-docs/security/mysql_backup_recovery_evidence.md. + +Developer: + Manish Kumar """ import subprocess from datetime import datetime, timedelta, timezone @@ -15,6 +18,9 @@ def _run(health_content: str | None, tmp_path: Path, max_age_hours: int = 30, require_encryption: bool | None = None) -> subprocess.CompletedProcess: + """Runs backup-health-check.sh with a minimal PATH and BACKUP_HEALTH_FILE pointing + at a synthetic file (or a nonexistent path when content is None), plus optional + max-age and require-encryption settings.""" env = {"PATH": "/usr/bin:/bin"} if health_content is not None: health_file = tmp_path / "health.env" @@ -33,12 +39,15 @@ def _iso(hours_ago: float) -> str: def test_missing_health_file_is_unhealthy(tmp_path): + """A missing health file makes the check exit 1 and report 'no health file'.""" result = _run(None, tmp_path) assert result.returncode == 1 assert "no health file" in result.stdout def test_recent_success_is_healthy(tmp_path): + """A success recorded an hour ago with an existing artifact exits 0 and reports + [OK].""" artifact = tmp_path / "fake.sql.gz" artifact.write_bytes(b"\x1f\x8b0") content = ( @@ -56,6 +65,8 @@ def test_recent_success_is_healthy(tmp_path): def test_backup_older_than_max_age_is_unhealthy(tmp_path): + """A last success 40 hours old against a 30-hour limit exits 1 and reports the + backup as old.""" artifact = tmp_path / "fake.sql.gz" artifact.write_bytes(b"\x1f\x8b0") content = ( @@ -73,6 +84,8 @@ def test_backup_older_than_max_age_is_unhealthy(tmp_path): def test_most_recent_failure_is_unhealthy_even_with_fresh_prior_success(tmp_path): + """A latest attempt that failed at the dump stage exits 1 with 'FAILED at stage', + even though an earlier success is only an hour old.""" artifact = tmp_path / "fake.sql.gz" artifact.write_bytes(b"\x1f\x8b0") content = ( @@ -90,6 +103,8 @@ def test_most_recent_failure_is_unhealthy_even_with_fresh_prior_success(tmp_path def test_missing_artifact_on_disk_is_unhealthy_even_if_recent(tmp_path): + """A recent success whose artifact file is gone exits 1 and reports that it no + longer exists.""" content = ( f"LAST_ATTEMPT_TS={_iso(1)}\n" "LAST_RESULT=success\n" @@ -123,11 +138,15 @@ def _healthy_content(tmp_path: Path, encrypted: str) -> str: def test_require_encryption_true_with_encrypted_backup_is_healthy(tmp_path): + """With encryption required, a backup recorded as LAST_ARTIFACT_ENCRYPTED=true exits + 0.""" result = _run(_healthy_content(tmp_path, "true"), tmp_path, require_encryption=True) assert result.returncode == 0, result.stdout def test_require_encryption_true_with_unencrypted_backup_is_unhealthy(tmp_path): + """With encryption required, a backup recorded as LAST_ARTIFACT_ENCRYPTED=false + exits 1 and reports 'encryption is required'.""" result = _run(_healthy_content(tmp_path, "false"), tmp_path, require_encryption=True) assert result.returncode == 1 assert "encryption is required" in result.stdout @@ -140,5 +159,7 @@ def test_require_encryption_false_with_unencrypted_backup_is_still_healthy(tmp_p def test_require_encryption_unset_does_not_assert_encryption(tmp_path): + """With BACKUP_REQUIRE_ENCRYPTION unset, an unencrypted backup is still healthy + because encryption is not asserted.""" result = _run(_healthy_content(tmp_path, "false"), tmp_path) assert result.returncode == 0, result.stdout diff --git a/tests/test_backup_mysql.py b/tests/test_backup_mysql.py index 3101d73d..22f39a0f 100644 --- a/tests/test_backup_mysql.py +++ b/tests/test_backup_mysql.py @@ -14,6 +14,9 @@ omnibioai-docs/security/mysql_backup_recovery_evidence.md -- that step is deliberately not mocked here, per the 2026-09-16 incident's closure requirements. + +Developer: + Manish Kumar """ import json import os @@ -54,6 +57,10 @@ class Sandbox: + """Isolated synthetic sandbox for running backup-mysql.sh: a throwaway backup dir, + health file and .env under tmp_path plus a fake docker shim on PATH (never the real + daemon), with configurable container state, mysqldump exit code and mysqldump + output.""" def __init__(self, tmp_path: Path, *, container_running=True, mysqldump_exit=0, mysqldump_output="-- fake sql dump\\nSELECT 1;\\n", env_content="MYSQL_ROOT_PASSWORD=test-password-not-real\\n"): self.tmp_path = tmp_path @@ -78,6 +85,8 @@ def __init__(self, tmp_path: Path, *, container_running=True, mysqldump_exit=0, fake_docker.chmod(fake_docker.stat().st_mode | stat.S_IEXEC) def run(self, extra_env: dict | None = None) -> subprocess.CompletedProcess: + """Runs backup-mysql.sh as a subprocess with PATH, OMNIBIOAI_ENV_FILE, + BACKUP_DIR and the health file redirected into the sandbox.""" env = os.environ.copy() env["PATH"] = f"{self.bin_dir}:{env['PATH']}" env["OMNIBIOAI_ENV_FILE"] = str(self.env_file) @@ -101,6 +110,8 @@ def partials(self): return sorted(self.backup_dir.glob("*.partial")) + sorted(self.backup_dir.glob("*.partial.gpg")) def health(self) -> dict: + """Parses the KEY=VALUE health file into a dict, or returns an empty dict when + the file does not exist.""" if not self.health_file.exists(): return {} out = {} @@ -113,11 +124,15 @@ def health(self) -> dict: @pytest.fixture def sandbox(tmp_path): + """Default sandbox: MySQL container running, mysqldump succeeding and a synthetic + .env with a placeholder password.""" return Sandbox(tmp_path) # 1. secret retrieval failure -> backup fails def test_missing_mysql_root_password_fails_closed(tmp_path): + """A .env with no MYSQL_ROOT_PASSWORD makes the backup exit non-zero, name the + missing variable on stderr and publish no artifact.""" sb = Sandbox(tmp_path, env_content="# no password set at all\n") result = sb.run() assert result.returncode != 0 @@ -127,6 +142,9 @@ def test_missing_mysql_root_password_fails_closed(tmp_path): # 13. malformed secret/config fails closed (not a crash with a stray artifact) def test_env_file_with_shell_metacharacter_secret_does_not_crash_and_loads(tmp_path): + """A password value containing parentheses (the shape of the original incident) + loads without a shell syntax error, and the backup succeeds with exactly one + artifact.""" # The exact incident shape: a password-shaped value containing parens. sb = Sandbox(tmp_path, env_content="MYSQL_ROOT_PASSWORD=abc(def)ghi\n") result = sb.run() @@ -137,6 +155,9 @@ def test_env_file_with_shell_metacharacter_secret_does_not_crash_and_loads(tmp_p # 2. database connection failure -> backup fails def test_container_not_running_fails_closed(tmp_path): + """When the MySQL container is not running, the backup fails with 'not running' on + stderr, publishes no artifact and records LAST_RESULT=failure at the preflight + stage.""" sb = Sandbox(tmp_path, container_running=False) result = sb.run() assert result.returncode != 0 @@ -149,6 +170,8 @@ def test_container_not_running_fails_closed(tmp_path): # 3. dump failure -> no successful artifact def test_mysqldump_failure_produces_no_artifact(tmp_path): + """A failing mysqldump leaves no artifact and no leftover .partial file, and the + health file records a failure at the dump stage.""" sb = Sandbox(tmp_path, mysqldump_exit=1) result = sb.run() assert result.returncode != 0 @@ -161,6 +184,8 @@ def test_mysqldump_failure_produces_no_artifact(tmp_path): # 6. partial backup never published as completed def test_failed_run_never_leaves_a_partial_at_the_final_name(tmp_path): + """After a failed dump, no file in the backup directory carries the final .sql.gz + name.""" sb = Sandbox(tmp_path, mysqldump_exit=1) sb.run() for f in sb.backup_dir.iterdir(): @@ -169,6 +194,8 @@ def test_failed_run_never_leaves_a_partial_at_the_final_name(tmp_path): # 7. success produces a completed artifact (with checksum sidecar) def test_successful_run_produces_completed_checksummed_artifact(tmp_path): + """A successful run publishes exactly one non-empty .sql.gz artifact together with a + .sha256 sidecar.""" sb = Sandbox(tmp_path) result = sb.run() assert result.returncode == 0, result.stderr @@ -181,6 +208,8 @@ def test_successful_run_produces_completed_checksummed_artifact(tmp_path): # 12. last-success state updates only after actual successful backup def test_last_success_timestamp_unchanged_by_a_failed_run(tmp_path): + """A later failed run records LAST_RESULT=failure but leaves LAST_SUCCESS_TS and + LAST_ARTIFACT from the earlier success untouched.""" sb = Sandbox(tmp_path) ok = sb.run() assert ok.returncode == 0, ok.stderr @@ -199,6 +228,8 @@ def test_last_success_timestamp_unchanged_by_a_failed_run(tmp_path): # 8. retention not run destructively after a failed backup def test_retention_never_runs_after_a_failed_backup(tmp_path): + """A run that fails before the dump (password removed) never reaches retention, so a + backdated existing artifact is not rotated away.""" sb = Sandbox(tmp_path) ok = sb.run() assert ok.returncode == 0, ok.stderr @@ -218,6 +249,7 @@ def test_retention_never_runs_after_a_failed_backup(tmp_path): # 9. retention failure is observable / retention never deletes the last backup def test_retention_never_deletes_the_last_remaining_backup(tmp_path): + """Even with RETAIN_DAYS=0, the backup just created survives retention.""" sb = Sandbox(tmp_path) ok = sb.run(extra_env={"RETAIN_DAYS": "0"}) assert ok.returncode == 0, ok.stderr @@ -226,6 +258,8 @@ def test_retention_never_deletes_the_last_remaining_backup(tmp_path): def test_retention_keeps_at_least_one_when_multiple_are_old(tmp_path): + """With two seeded old artifacts and RETAIN_DAYS=7, at least one artifact remains + and the freshly created one survives.""" sb = Sandbox(tmp_path) # Seed two fake old artifacts directly (faster than 3 real runs). old_time = 1_000_000 @@ -244,6 +278,8 @@ def test_retention_keeps_at_least_one_when_multiple_are_old(tmp_path): # 10. secret values not emitted to logs def test_password_value_never_appears_in_script_output(tmp_path): + """The MYSQL_ROOT_PASSWORD marker value appears in neither stdout, stderr, the + health file nor any .sha256 sidecar.""" secret_marker = "SUPER-SECRET-MARKER-VALUE-9f8e7d" sb = Sandbox(tmp_path, env_content=f"MYSQL_ROOT_PASSWORD={secret_marker}\n") result = sb.run() @@ -261,6 +297,8 @@ def test_password_value_never_appears_in_script_output(tmp_path): # asserts via result.returncode != 0 -- restated explicitly here since it's # the thing that actually made the real incident invisible to cron/logs). def test_exit_code_propagates_on_every_failure_path(tmp_path): + """Each failure mode (container down, mysqldump failure, missing password) makes the + script exit non-zero, so a scheduler sees the failure.""" for i, kwargs in enumerate(( dict(container_running=False), dict(mysqldump_exit=1), @@ -277,6 +315,8 @@ def test_exit_code_propagates_on_every_failure_path(tmp_path): # Track E4 — backup encryption at rest (real gpg, never faked) # ============================================================ def _make_passphrase_file(tmp_path: Path, content: str = "correct-horse-battery-staple-not-real") -> Path: + """Writes a throwaway 0600 passphrase file with a synthetic, non-real value and + returns its path.""" f = tmp_path / "passphrase" f.write_text(content, encoding="utf-8") f.chmod(0o600) @@ -308,6 +348,8 @@ def test_passphrase_file_configured_encrypts_automatically(tmp_path): def test_encrypted_backup_is_real_ciphertext_not_gzip(tmp_path): + """With a passphrase file configured, the published artifact does not begin with the + gzip magic bytes, i.e. it is gpg output rather than plain gzip.""" sb = Sandbox(tmp_path) passphrase_file = _make_passphrase_file(tmp_path) result = sb.run(extra_env={"MYSQL_BACKUP_ENCRYPTION_PASSPHRASE_FILE": str(passphrase_file)}) @@ -342,6 +384,8 @@ def test_real_encrypt_then_decrypt_round_trip_recovers_original_dump(tmp_path): def test_wrong_passphrase_fails_closed_on_decrypt(tmp_path): + """Decrypting an encrypted backup with a different passphrase exits non-zero and + leaves no non-empty output file.""" dump_body = "-- fake sql dump\\nSELECT 1;\\n" sb = Sandbox(tmp_path, mysqldump_output=dump_body) passphrase_file = _make_passphrase_file(tmp_path, content="right-passphrase") @@ -363,6 +407,8 @@ def test_wrong_passphrase_fails_closed_on_decrypt(tmp_path): def test_corrupted_ciphertext_fails_closed_on_decrypt(tmp_path): + """Decrypting ciphertext whose mid-stream bytes were flipped exits non-zero instead + of producing output.""" sb = Sandbox(tmp_path) passphrase_file = _make_passphrase_file(tmp_path) result = sb.run(extra_env={"MYSQL_BACKUP_ENCRYPTION_PASSPHRASE_FILE": str(passphrase_file)}) @@ -388,6 +434,8 @@ def test_corrupted_ciphertext_fails_closed_on_decrypt(tmp_path): def test_require_encryption_true_without_passphrase_file_fails_closed(tmp_path): + """MYSQL_BACKUP_REQUIRE_ENCRYPTION=true with no passphrase file fails naming the + missing variable and leaves no plaintext, encrypted or .partial artifact.""" sb = Sandbox(tmp_path) result = sb.run(extra_env={"MYSQL_BACKUP_REQUIRE_ENCRYPTION": "true"}) assert result.returncode != 0 @@ -398,6 +446,8 @@ def test_require_encryption_true_without_passphrase_file_fails_closed(tmp_path): def test_missing_passphrase_file_fails_closed_no_plaintext_fallback(tmp_path): + """A configured but nonexistent passphrase file fails with 'passphrase file missing + or empty' and never falls back to a plaintext artifact.""" sb = Sandbox(tmp_path) nonexistent = tmp_path / "does-not-exist-passphrase" result = sb.run(extra_env={"MYSQL_BACKUP_ENCRYPTION_PASSPHRASE_FILE": str(nonexistent)}) @@ -409,6 +459,8 @@ def test_missing_passphrase_file_fails_closed_no_plaintext_fallback(tmp_path): def test_empty_passphrase_file_fails_closed(tmp_path): + """An empty passphrase file fails the run and publishes neither a plaintext nor an + encrypted artifact.""" sb = Sandbox(tmp_path) empty = tmp_path / "empty-passphrase" empty.write_text("", encoding="utf-8") @@ -458,6 +510,8 @@ def test_checksum_covers_the_ciphertext_not_the_plaintext(tmp_path): def test_passphrase_file_path_ok_but_contents_never_appear_in_logs(tmp_path): + """The passphrase file's contents appear in neither stdout, stderr nor the health + file.""" secret_marker = "SUPER-SECRET-PASSPHRASE-MARKER-a1b2c3" sb = Sandbox(tmp_path) passphrase_file = _make_passphrase_file(tmp_path, content=secret_marker) @@ -469,6 +523,8 @@ def test_passphrase_file_path_ok_but_contents_never_appear_in_logs(tmp_path): def test_encryption_status_recorded_in_health_file_and_preserved_on_later_failure(tmp_path): + """LAST_ARTIFACT_ENCRYPTED=true is recorded after an encrypted success and is + preserved when a later run fails.""" sb = Sandbox(tmp_path) passphrase_file = _make_passphrase_file(tmp_path) ok = sb.run(extra_env={"MYSQL_BACKUP_ENCRYPTION_PASSPHRASE_FILE": str(passphrase_file)}) @@ -483,6 +539,8 @@ def test_encryption_status_recorded_in_health_file_and_preserved_on_later_failur def test_backup_failure_emits_a_backup_failed_security_alert(tmp_path): + """A failed backup prints a [SECURITY-ALERT] JSON line with condition backup_failed, + severity critical and component mysql-backup.""" sb = Sandbox(tmp_path, container_running=False) result = sb.run() assert result.returncode != 0 @@ -495,6 +553,8 @@ def test_backup_failure_emits_a_backup_failed_security_alert(tmp_path): def test_encryption_stage_failure_emits_a_distinct_condition(tmp_path): + """A gpg failure emits a security alert with condition backup_encryption_failed, + distinguishable from the generic backup_failed.""" sb = Sandbox(tmp_path) passphrase_file = _make_passphrase_file(tmp_path) fake_gpg = sb.bin_dir / "gpg" @@ -510,6 +570,8 @@ def test_encryption_stage_failure_emits_a_distinct_condition(tmp_path): def test_repeated_backup_failures_are_deduped_not_stormed(tmp_path): + """A second identical failure within the dedup window emits no further + [SECURITY-ALERT] line.""" sb = Sandbox(tmp_path, container_running=False) r1 = sb.run() r2 = sb.run() @@ -518,6 +580,7 @@ def test_repeated_backup_failures_are_deduped_not_stormed(tmp_path): def test_successful_backup_emits_no_alert(tmp_path): + """A successful backup emits no [SECURITY-ALERT] line.""" sb = Sandbox(tmp_path) result = sb.run() assert result.returncode == 0, result.stderr @@ -525,6 +588,8 @@ def test_successful_backup_emits_no_alert(tmp_path): def test_retention_rotates_both_plaintext_and_encrypted_artifacts(tmp_path): + """Retention removes old artifacts of both kinds (.sql.gz and .sql.gz.gpg), leaving + only the newly created encrypted artifact.""" sb = Sandbox(tmp_path) old_time = 1_000_000 plain = sb.backup_dir / "omnibioai_20200101_000000.sql.gz" diff --git a/tests/test_canonical_compose_path.py b/tests/test_canonical_compose_path.py index 3953e12e..a780ace2 100644 --- a/tests/test_canonical_compose_path.py +++ b/tests/test_canonical_compose_path.py @@ -1,3 +1,11 @@ +"""Canonical Compose path guards: the Settings page defaults to the root +docker-compose.yml, and the legacy docker/docker-compose.yml heredoc no longer exists as +a Compose file. + +Developer: + Manish Kumar +""" + from pathlib import Path @@ -5,10 +13,14 @@ def test_settings_default_uses_canonical_root_compose(): + """Settings.jsx defaults compose_file to the root docker-compose.yml and never + references docker/docker-compose.yml.""" settings = (ROOT / "src/ui/pages/Settings.jsx").read_text(encoding="utf-8") assert 'compose_file: "docker-compose.yml"' in settings assert '"docker/docker-compose.yml"' not in settings def test_legacy_heredoc_is_not_presented_as_compose_yaml(): + """The legacy docker/docker-compose.yml path does not exist on disk, so it cannot be + mistaken for a real Compose file.""" assert not (ROOT / "docker/docker-compose.yml").exists() diff --git a/tests/test_check_env.py b/tests/test_check_env.py index f3d77f61..a6746373 100644 --- a/tests/test_check_env.py +++ b/tests/test_check_env.py @@ -5,6 +5,9 @@ validator would itself crash on the same rotated secret it exists to help catch problems with. See scripts/lib-env.sh and omnibioai-docs/security/mysql_backup_recovery_evidence.md. + +Developer: + Manish Kumar """ import subprocess import tempfile @@ -14,6 +17,8 @@ def test_uses_the_safe_loader_not_the_vulnerable_pattern(): + """check-env.sh loads its .env through lib-env.sh's load_env_file and its + non-comment code never uses the source <( ... ) pattern.""" text = SCRIPT.read_text(encoding="utf-8") code_lines = [ln for ln in text.splitlines() if not ln.strip().startswith("#")] code = "\n".join(code_lines) @@ -23,6 +28,9 @@ def test_uses_the_safe_loader_not_the_vulnerable_pattern(): def test_survives_a_value_containing_parentheses(): + """check-env.sh exits 0 and reports all critical secrets set when a secret value + contains parentheses and shell metacharacters; run against a throwaway copy in a + temp dir.""" with tempfile.TemporaryDirectory() as td: root = Path(td) (root / "scripts").mkdir() diff --git a/tests/test_compose_network_exposure.py b/tests/test_compose_network_exposure.py index e0b58f60..d6ab0ca5 100644 --- a/tests/test_compose_network_exposure.py +++ b/tests/test_compose_network_exposure.py @@ -28,6 +28,9 @@ Development access to mysql/redis is preserved through an explicit, separate overlay file rather than by weakening the release default -- see SECURITY-COMPOSE-HARDENING.md. + +Developer: + Manish Kumar """ from pathlib import Path @@ -89,6 +92,8 @@ def _config_only(path): scope="module", params=RELEASE_COMPOSE_PATHS, ids=lambda p: p.name ) def release_compose(request): + """Parses each release compose file (parametrized over the dot and dash variants) + once per module and returns the loaded YAML.""" return _load(request.param) @@ -96,6 +101,8 @@ def release_compose(request): scope="module", params=RELEASE_COMPOSE_PATHS, ids=lambda p: p.name ) def release_compose_text(request): + """Returns each release compose file's text with comment lines removed, so substring + checks see configuration rather than prose that quotes removed defaults.""" return _config_only(request.param) @@ -276,6 +283,8 @@ def test_api_gateway_remains_published(release_compose): def test_dev_ports_overlay_exists(): + """The development-only overlay docker-compose.release.dev-ports.yml exists, so + local datastore access does not require editing the release files.""" assert DEV_PORTS_OVERLAY.exists(), ( "an explicit development-only overlay must exist so local datastore " "access is achievable without weakening the release default" @@ -360,6 +369,8 @@ def test_dev_compose_still_publishes_for_local_development(): @pytest.mark.parametrize("compose_path", ALL_COMPOSE_PATHS, ids=lambda p: p.name) def test_redis_has_aof_durability_enabled(compose_path): + """Parametrized over the dev compose file and both release files: the redis command + enables AOF (--appendonly yes) with an explicit --appendfsync everysec policy.""" config = _load(compose_path) command = config["services"]["redis"].get("command", "") assert "--appendonly yes" in command, ( @@ -374,12 +385,9 @@ def test_redis_has_aof_durability_enabled(compose_path): def test_redis_aof_config_is_identical_across_all_profiles(): - """The two release files (docker-compose.release.yml and the - dash-named legacy docker-compose-release.yml) must be kept in parity - -- this exact drift (one file getting a fix, the other silently not) - is what test_dev_ports_overlay_matches_release_baseline and friends - already guard against for other settings; this pins it for the AOF - command specifically.""" + """The redis service command must be identical across docker-compose.yml and both + release files (docker-compose.release.yml and docker-compose-release.yml), so a + redis-server flag change cannot land in only one profile.""" commands = { p.name: _load(p)["services"]["redis"].get("command", "") for p in ALL_COMPOSE_PATHS diff --git a/tests/test_compose_release_config.py b/tests/test_compose_release_config.py index 2d8740e9..813bd3de 100644 --- a/tests/test_compose_release_config.py +++ b/tests/test_compose_release_config.py @@ -14,6 +14,9 @@ installer -- the dash file was fixed by PR1/PR2 while the dot file silently kept shipping the unfixed gap. Now parametrized over both files so neither can drift out of sync with the other again. + +Developer: + Manish Kumar """ from pathlib import Path @@ -71,12 +74,17 @@ @pytest.fixture(scope="module", params=COMPOSE_PATHS, ids=lambda p: p.name) def compose_config(request): + """Parses each release compose file (parametrized over the dash and dot variants) + once per module and returns the loaded YAML.""" with open(request.param) as f: return yaml.safe_load(f) @pytest.mark.parametrize("service,env_key", EXPECTED_SECRET_WIRING.items()) def test_service_receives_shared_auth_secret(compose_config, service, env_key): + """Parametrized over every JWT consumer in EXPECTED_SECRET_WIRING: the service + defines its secret env key in each release compose file, and the value is sourced + from AUTH_SECRET_KEY rather than hardcoded.""" env = compose_config["services"][service]["environment"] assert env_key in env, ( f"{service} must receive {env_key} -- without it, this service's " diff --git a/tests/test_integration_credentials.py b/tests/test_integration_credentials.py index 48efa6dd..be67345a 100644 --- a/tests/test_integration_credentials.py +++ b/tests/test_integration_credentials.py @@ -1,3 +1,10 @@ +"""Static guard that the live-integration conftest never ships literal credential +defaults. The file is parsed with ast and is never imported or executed. + +Developer: + Manish Kumar +""" + import ast from pathlib import Path @@ -13,6 +20,9 @@ def test_integration_credentials_have_no_nonempty_literal_defaults(): + """Each integration credential in tests/integration/conftest.py is read with + os.getenv and an empty-string default, so no non-empty literal credential can be + committed.""" tree = ast.parse(CONFTEST.read_text(encoding="utf-8")) assignments = { node.targets[0].id: node.value diff --git a/tests/test_lib_alert.py b/tests/test_lib_alert.py index f99a15b4..590a5cfa 100644 --- a/tests/test_lib_alert.py +++ b/tests/test_lib_alert.py @@ -2,6 +2,9 @@ security alert emission, matching omnibioai-security-audit's audit/security_alerts.py schema. Tests invoke the real bash function via a tiny wrapper script, never mocked. + +Developer: + Manish Kumar """ import json import shlex @@ -12,6 +15,8 @@ def _run(component, condition, severity, message, *kv_pairs, env_overrides=None, tmp_path=None): + """Runs emit_security_alert from lib-alert.sh in a real bash subprocess with a + minimal PATH and a per-test dedup state directory under tmp_path.""" state_dir = tmp_path / "alert-state" if tmp_path else None args = [component, condition, severity, message, *kv_pairs] script = f""" @@ -28,11 +33,15 @@ def _run(component, condition, severity, message, *kv_pairs, env_overrides=None, def _parse_alert_line(stdout: str) -> dict: + """Extracts and JSON-decodes the first [SECURITY-ALERT] line from the script's + stdout.""" line = next(line for line in stdout.splitlines() if line.startswith("[SECURITY-ALERT] ")) return json.loads(line[len("[SECURITY-ALERT] "):]) def test_emits_a_json_line_with_the_expected_schema(tmp_path): + """emit_security_alert prints one [SECURITY-ALERT] JSON line carrying condition, + severity, component, message, metadata and a timestamp.""" result = _run("mysql-backup", "backup_failed", "critical", "dump failed", "stage=dump", tmp_path=tmp_path) assert result.returncode == 0, result.stderr alert = _parse_alert_line(result.stdout) @@ -45,18 +54,22 @@ def test_emits_a_json_line_with_the_expected_schema(tmp_path): def test_no_metadata_args_produces_empty_object(tmp_path): + """With no key=value arguments, the alert's metadata is an empty object.""" result = _run("comp", "cond", "warning", "msg", tmp_path=tmp_path) alert = _parse_alert_line(result.stdout) assert alert["metadata"] == {} def test_multiple_metadata_pairs(tmp_path): + """Multiple key=value arguments become string entries in the alert's metadata.""" result = _run("comp", "cond", "warning", "msg", "a=1", "b=two", tmp_path=tmp_path) alert = _parse_alert_line(result.stdout) assert alert["metadata"] == {"a": "1", "b": "two"} def test_repeated_identical_condition_is_deduped_within_window(tmp_path): + """A repeat of the same condition within the dedup window is suppressed, and the + dedup state directory is created.""" state_dir = tmp_path / "alert-state" r1 = _run("comp", "cond", "warning", "msg", tmp_path=tmp_path) r2 = _run("comp", "cond", "warning", "msg", tmp_path=tmp_path) @@ -66,6 +79,7 @@ def test_repeated_identical_condition_is_deduped_within_window(tmp_path): def test_different_conditions_are_not_deduped_against_each_other(tmp_path): + """Alerts with different conditions do not suppress one another.""" r1 = _run("comp", "cond-a", "warning", "msg", tmp_path=tmp_path) r2 = _run("comp", "cond-b", "warning", "msg", tmp_path=tmp_path) assert "[SECURITY-ALERT]" in r1.stdout @@ -73,6 +87,7 @@ def test_different_conditions_are_not_deduped_against_each_other(tmp_path): def test_zero_window_allows_immediate_repeat(tmp_path): + """BACKUP_ALERT_DEDUP_WINDOW_SECONDS=0 lets an identical alert repeat immediately.""" r1 = _run("comp", "cond", "warning", "msg", tmp_path=tmp_path, env_overrides={"BACKUP_ALERT_DEDUP_WINDOW_SECONDS": "0"}) r2 = _run("comp", "cond", "warning", "msg", tmp_path=tmp_path, env_overrides={"BACKUP_ALERT_DEDUP_WINDOW_SECONDS": "0"}) assert "[SECURITY-ALERT]" in r1.stdout @@ -80,6 +95,8 @@ def test_zero_window_allows_immediate_repeat(tmp_path): def test_recovered_condition_bypasses_dedup(tmp_path): + """After a repeat of restore_verification_failed is deduped, a + restore_verification_passed_recovered alert is still emitted.""" _run("comp", "restore_verification_failed", "critical", "msg", tmp_path=tmp_path) r2 = _run("comp", "restore_verification_failed", "critical", "msg", tmp_path=tmp_path) # deduped r3 = _run("comp", "restore_verification_passed_recovered", "info", "recovered", tmp_path=tmp_path) @@ -88,6 +105,7 @@ def test_recovered_condition_bypasses_dedup(tmp_path): def test_message_with_special_characters_produces_valid_json(tmp_path): + """A message containing double quotes and a backslash round-trips as valid JSON.""" result = _run("comp", "cond", "warning", 'a "quoted" message with \\ backslash', tmp_path=tmp_path) alert = _parse_alert_line(result.stdout) assert alert["message"] == 'a "quoted" message with \\ backslash' @@ -110,6 +128,8 @@ def test_unwritable_state_dir_still_emits_the_alert(tmp_path): def test_optional_log_file_receives_the_same_jsonl(tmp_path): + """With BACKUP_ALERT_LOG_FILE set, the same alert is also written as exactly one + JSON line to that file.""" log_file = tmp_path / "alerts.jsonl" result = _run("comp", "cond", "warning", "msg", "k=v", tmp_path=tmp_path, env_overrides={"BACKUP_ALERT_LOG_FILE": str(log_file)}) diff --git a/tests/test_lib_env.py b/tests/test_lib_env.py index 8d96e722..7fcd1ca7 100644 --- a/tests/test_lib_env.py +++ b/tests/test_lib_env.py @@ -16,6 +16,9 @@ Real subprocess execution against synthetic, throwaway .env content -- no real secrets, no docker, no network. Values below are placeholders invented for this test file only. + +Developer: + Manish Kumar """ import subprocess import tempfile @@ -25,6 +28,9 @@ def _load_and_echo(env_content: str, var_names: list[str]) -> subprocess.CompletedProcess: + """Writes env_content to a throwaway .env, loads it with load_env_file in a real + bash subprocess (set -euo pipefail, allexport) and echoes each requested variable, + or __UNSET__ when it was not loaded.""" with tempfile.TemporaryDirectory() as td: env_file = Path(td) / ".env" env_file.write_text(env_content, encoding="utf-8") @@ -43,6 +49,8 @@ def _load_and_echo(env_content: str, var_names: list[str]) -> subprocess.Complet def test_value_with_parentheses_loads_without_syntax_error(): + """An unquoted value containing parentheses (the shape of the original incident) + loads with exit 0, no bash syntax error and the value preserved verbatim.""" # The exact shape of the real incident: an unquoted value containing "(" and ")". result = _load_and_echo("SECRET_KEY=abc(def)ghi\n", ["SECRET_KEY"]) assert result.returncode == 0, result.stderr @@ -51,12 +59,15 @@ def test_value_with_parentheses_loads_without_syntax_error(): def test_value_with_dollar_ampersand_asterisk_bang_hash_loads_literally(): + """A value containing $, &, *, ! and # characters is loaded as literal text.""" result = _load_and_echo("SECRET_KEY=a$$b&c*d!e#f\n", ["SECRET_KEY"]) assert result.returncode == 0, result.stderr assert "SECRET_KEY=a$$b&c*d!e#f" in result.stdout def test_value_with_backtick_does_not_trigger_command_substitution(): + """A value containing backticks is loaded as literal text and not executed as + command substitution.""" result = _load_and_echo("SECRET_KEY=`whoami`\n", ["SECRET_KEY"]) assert result.returncode == 0, result.stderr # Must load the literal text "`whoami`", not the output of running whoami. @@ -65,24 +76,29 @@ def test_value_with_backtick_does_not_trigger_command_substitution(): def test_value_containing_quotes_loads_literally(): + """A value containing both single and double quotes is loaded literally.""" result = _load_and_echo('SECRET_KEY=it\'s "quoted" text\n', ["SECRET_KEY"]) assert result.returncode == 0, result.stderr assert """SECRET_KEY=it's "quoted" text""" in result.stdout def test_comments_and_blank_lines_are_skipped(): + """Comment lines and blank lines are skipped while later assignments still load.""" result = _load_and_echo("# a comment\n\nFOO=bar\n", ["FOO"]) assert result.returncode == 0, result.stderr assert "FOO=bar" in result.stdout def test_export_prefixed_line_is_handled(): + """A line prefixed with export is parsed as an ordinary assignment.""" result = _load_and_echo("export FOO=bar\n", ["FOO"]) assert result.returncode == 0, result.stderr assert "FOO=bar" in result.stdout def test_missing_env_file_is_not_fatal(): + """A nonexistent env file is not fatal: the loader returns and the calling script + continues.""" script = f''' set -euo pipefail source "{LIB}" @@ -95,6 +111,8 @@ def test_missing_env_file_is_not_fatal(): def test_malformed_line_without_equals_is_skipped_not_fatal(): + """A line with no equals sign is skipped without error and later well-formed lines + still load.""" # Fail-closed-but-not-crashing: a garbled line is ignored, later # well-formed lines still load, and the loader itself never errors. result = _load_and_echo("THIS_LINE_HAS_NO_EQUALS_SIGN\nFOO=bar\n", ["FOO"]) @@ -103,12 +121,15 @@ def test_malformed_line_without_equals_is_skipped_not_fatal(): def test_value_containing_equals_signs_is_preserved_in_full(): + """A value that itself contains equals signs is preserved in full.""" result = _load_and_echo("SECRET_KEY=a=b=c\n", ["SECRET_KEY"]) assert result.returncode == 0, result.stderr assert "SECRET_KEY=a=b=c" in result.stdout def test_loader_never_uses_source_eval_or_process_substitution(): + """The loader's non-comment code contains no source <( pattern and no eval, guarding + against a return to the vulnerable loader.""" # Structural guard against regressing back to the vulnerable pattern. # Only the code matters here -- the header comment deliberately quotes # the old broken pattern for incident context, so strip comment lines diff --git a/tests/test_lims_debug_config.py b/tests/test_lims_debug_config.py index f05c14a7..dc23962e 100644 --- a/tests/test_lims_debug_config.py +++ b/tests/test_lims_debug_config.py @@ -25,6 +25,9 @@ suite pins both files so that specific drift can't silently reappear -- mirroring test_compose_release_config.py's approach to the JWT-secret drift between the same two files. + +Developer: + Manish Kumar """ from pathlib import Path @@ -51,6 +54,8 @@ def _load(path): @pytest.fixture(scope="module", params=RELEASE_COMPOSE_PATHS, ids=lambda p: p.name) def release_compose(request): + """Loads each release compose file (parametrized over the dot and dash variants) for + the lims DEBUG and encryption-key checks.""" return _load(request.param) diff --git a/tests/test_mysql_recovery_drill.py b/tests/test_mysql_recovery_drill.py index 80e85e78..916c98d5 100644 --- a/tests/test_mysql_recovery_drill.py +++ b/tests/test_mysql_recovery_drill.py @@ -1,3 +1,11 @@ +"""Static guards on scripts/mysql-recovery-drill.sh: the recovery drill must stay +isolated from the live MySQL container and use only synthetic data. The script text is +inspected and never executed. + +Developer: + Manish Kumar +""" + from pathlib import Path @@ -5,6 +13,9 @@ def test_drill_is_isolated_and_uses_only_synthetic_data(): + """The script text uses --network none, an empty-password throwaway MySQL and + synthetic readiness_drill data, and never mentions the live omnibioai-studio-mysql-1 + container or any .env file.""" text = SCRIPT.read_text(encoding="utf-8") assert "--network none" in text assert "MYSQL_ALLOW_EMPTY_PASSWORD=yes" in text @@ -15,6 +26,8 @@ def test_drill_is_isolated_and_uses_only_synthetic_data(): def test_drill_has_cleanup_and_integrity_checks(): + """The script text installs an EXIT cleanup trap and contains the index_checks=PASS, + constraint_checks=PASS and rows_after_write integrity checks.""" text = SCRIPT.read_text(encoding="utf-8") assert "trap cleanup EXIT" in text assert "index_checks=PASS" in text diff --git a/tests/test_safe_compose_diagnostics.py b/tests/test_safe_compose_diagnostics.py index d2cf4e16..10add3bf 100644 --- a/tests/test_safe_compose_diagnostics.py +++ b/tests/test_safe_compose_diagnostics.py @@ -1,3 +1,11 @@ +"""scripts/safe_compose_diagnostics.py: the diagnostics report may show environment +variable names and SET/UNSET presence, but must not print values taken from .env or from +Compose interpolation defaults. + +Developer: + Manish Kumar +""" + from pathlib import Path from scripts.safe_compose_diagnostics import dotenv_names, load_compose, report @@ -7,6 +15,8 @@ def test_report_exposes_names_and_presence_but_never_values(tmp_path: Path): + """The report marks env names SET or UNSET from the .env name set, and neither a + .env value nor a redis:// URL appears in the output.""" compose_path = tmp_path / "compose.yml" env_path = tmp_path / ".env" compose_path.write_text( @@ -36,6 +46,8 @@ def test_report_exposes_names_and_presence_but_never_values(tmp_path: Path): def test_report_does_not_resolve_interpolation_default(tmp_path: Path): + """A Compose interpolation default such as ${API_TOKEN:-} is reported as + UNSET and the default value is never echoed.""" compose_path = tmp_path / "compose.yml" compose_path.write_text( f"services:\n api:\n environment:\n API_TOKEN: ${{API_TOKEN:-{SECRET_SENTINEL}}}\n", diff --git a/tests/test_service_catalog_drift.py b/tests/test_service_catalog_drift.py index 22d3bfcd..89f2ac3a 100644 --- a/tests/test_service_catalog_drift.py +++ b/tests/test_service_catalog_drift.py @@ -1,9 +1,18 @@ +"""Drift guard between the services declared in docker-compose.yml and the service +catalog in docs/SYSTEM_ARCHITECTURE.md. + +Developer: + Manish Kumar +""" + from pathlib import Path from scripts.check_service_catalog import catalog_services, compose_services, duplicates def test_compose_and_catalog_match(): + """Compose and the architecture catalog each list exactly 41 services, the two + service sets are identical, and the catalog has no duplicate entries.""" root = Path(__file__).resolve().parents[1] compose = compose_services(root / "docker-compose.yml") catalog = catalog_services(root / "docs" / "SYSTEM_ARCHITECTURE.md") diff --git a/tests/test_verify_mysql_backup_restore.py b/tests/test_verify_mysql_backup_restore.py index 8863b7af..f7a456cb 100644 --- a/tests/test_verify_mysql_backup_restore.py +++ b/tests/test_verify_mysql_backup_restore.py @@ -6,6 +6,9 @@ slow and already exercised manually -- see omnibioai-docs/security/mysql_backup_recovery_evidence.md for the real, non-mocked restore-proof run against an actual backup artifact). + +Developer: + Manish Kumar """ from pathlib import Path @@ -13,6 +16,9 @@ def test_restore_target_is_network_isolated_and_disposable(): + """The script text starts the restore target with --network none and an + empty-password throwaway MySQL, installs an EXIT cleanup trap and removes the + container with docker rm -f.""" text = SCRIPT.read_text(encoding="utf-8") assert "--network none" in text assert "MYSQL_ALLOW_EMPTY_PASSWORD=yes" in text @@ -21,17 +27,22 @@ def test_restore_target_is_network_isolated_and_disposable(): def test_never_targets_the_real_production_container(): + """The script text never mentions the live omnibioai-studio-mysql-1 container.""" text = SCRIPT.read_text(encoding="utf-8") assert "omnibioai-studio-mysql-1" not in text def test_verifies_integrity_before_restoring(): + """The script text contains both the gzip -t and sha256sum -c integrity checks; only + their presence is asserted here, not their order.""" text = SCRIPT.read_text(encoding="utf-8") assert "gzip -t" in text assert "sha256sum -c" in text def test_compares_declared_vs_restored_structure(): + """The script text creates databases and compares information_schema tables and + schemata, i.e. declared versus restored structure.""" text = SCRIPT.read_text(encoding="utf-8") assert "CREATE DATABASE" in text assert "information_schema.tables" in text @@ -39,6 +50,8 @@ def test_compares_declared_vs_restored_structure(): def test_row_content_is_never_selected_only_counts(): + """The script text uses SELECT COUNT(*) row counts and never SELECT *, so row + contents are not read out.""" text = SCRIPT.read_text(encoding="utf-8") assert "SELECT COUNT(*)" in text assert "SELECT *" not in text @@ -48,6 +61,9 @@ def test_row_content_is_never_selected_only_counts(): # Track E4 — encrypted-artifact decrypt path # ============================================================ def test_decrypts_gpg_artifacts_and_checks_exit_code_explicitly(): + """The script text branches on .gpg artifacts, runs gpg --batch and checks its exit + status with an explicit 'if ! gpg' and a 'gpg decryption failed' message instead of + relying on set -e.""" text = SCRIPT.read_text(encoding="utf-8") assert '"${ARTIFACT}" == *.gpg' in text assert "gpg --batch" in text @@ -59,6 +75,8 @@ def test_decrypts_gpg_artifacts_and_checks_exit_code_explicitly(): def test_decrypt_requires_passphrase_file_env_var_fails_closed_if_unset(): + """The script text references MYSQL_BACKUP_ENCRYPTION_PASSPHRASE_FILE and contains + an 'is not set' failure message for the unset case.""" text = SCRIPT.read_text(encoding="utf-8") assert "MYSQL_BACKUP_ENCRYPTION_PASSPHRASE_FILE" in text assert "is not set" in text @@ -95,6 +113,8 @@ def test_waits_for_temporary_server_to_stop_before_pinging(): def test_readiness_waits_are_bounded_not_infinite(): + """The script text bounds its readiness waits with MYSQL_INIT_WAIT_SECONDS and + MYSQL_READY_WAIT_SECONDS.""" text = SCRIPT.read_text(encoding="utf-8") assert "MYSQL_INIT_WAIT_SECONDS" in text assert "MYSQL_READY_WAIT_SECONDS" in text @@ -111,6 +131,8 @@ def test_restore_command_failure_is_explicitly_checked(): def test_sources_the_shared_alert_library_and_uses_a_fail_helper(): + """The script text sources lib-alert.sh and reports restore_verification_failed + through emit_security_alert.""" text = SCRIPT.read_text(encoding="utf-8") assert "lib-alert.sh" in text assert "emit_security_alert" in text @@ -127,6 +149,8 @@ def test_every_early_failure_path_uses_fail_not_bare_exit(): def test_verifies_e3_audit_hardening_structures_after_restore(): + """The script text checks the restored audit hardening structures: + record_integrity_hash, audit_legal_holds and information_schema.triggers.""" text = SCRIPT.read_text(encoding="utf-8") assert "record_integrity_hash" in text assert "audit_legal_holds" in text @@ -145,6 +169,8 @@ def test_e3_structure_check_targets_the_real_hardened_database_only(): def test_append_only_trigger_is_functionally_probed_not_just_checked_for_existence(): + """The script text functionally probes the append-only trigger with an UPDATE using + the e4-restore-probe marker, rather than only checking that the trigger exists.""" text = SCRIPT.read_text(encoding="utf-8") assert "UPDATE" in text assert "e4-restore-probe" in text diff --git a/tests/unit/test_license_server.py b/tests/unit/test_license_server.py index 0d82ea75..8c078304 100644 --- a/tests/unit/test_license_server.py +++ b/tests/unit/test_license_server.py @@ -6,6 +6,9 @@ pymysql.connect is patched before importing license_server, since that module calls init_db() (a real MySQL connection attempt) at import time. + +Developer: + Manish Kumar """ import sys from pathlib import Path @@ -23,10 +26,14 @@ @pytest.fixture def client(): + """FastAPI TestClient bound to license_server.app; pymysql was already patched when + the module was imported.""" return TestClient(license_server.app) def _mock_response(status_code=200, json_data=None): + """Builds a MagicMock HTTP response with the given status_code and a json() + returning json_data, or an empty dict when none is given.""" resp = MagicMock() resp.status_code = status_code resp.json.return_value = json_data or {} @@ -37,6 +44,8 @@ def _mock_response(status_code=200, json_data=None): def test_validate_token_returns_user_dict_on_valid_token(monkeypatch): + """_validate_token returns the user dict (email and permissions included) when the + mocked auth service reports the token valid.""" monkeypatch.setattr( license_server.requests, "post", lambda *a, **kw: _mock_response(200, { @@ -50,6 +59,7 @@ def test_validate_token_returns_user_dict_on_valid_token(monkeypatch): def test_validate_token_returns_none_on_invalid_token(monkeypatch): + """_validate_token returns None when the mocked auth service reports valid=False.""" monkeypatch.setattr( license_server.requests, "post", lambda *a, **kw: _mock_response(200, {"valid": False}), ) @@ -57,6 +67,8 @@ def test_validate_token_returns_none_on_invalid_token(monkeypatch): def test_validate_token_returns_none_on_network_error(monkeypatch): + """_validate_token returns None instead of raising when the auth service is + unreachable.""" def _boom(*a, **kw): raise ConnectionError("auth service unreachable") monkeypatch.setattr(license_server.requests, "post", _boom) @@ -67,18 +79,22 @@ def _boom(*a, **kw): def test_require_permission_missing_header_raises_401(): + """_require_permission raises HTTP 401 when no Authorization header is supplied.""" with pytest.raises(license_server.HTTPException) as exc_info: license_server._require_permission(None, "manage_licenses", action="test") assert exc_info.value.status_code == 401 def test_require_permission_non_bearer_header_raises_401(): + """_require_permission raises HTTP 401 for a non-Bearer Authorization scheme + (Basic).""" with pytest.raises(license_server.HTTPException) as exc_info: license_server._require_permission("Basic abc123", "manage_licenses", action="test") assert exc_info.value.status_code == 401 def test_require_permission_invalid_token_raises_401(monkeypatch): + """_require_permission raises HTTP 401 when token validation returns None.""" monkeypatch.setattr(license_server, "_validate_token", lambda token: None) with pytest.raises(license_server.HTTPException) as exc_info: license_server._require_permission("Bearer badtoken", "manage_licenses", action="test") @@ -86,6 +102,8 @@ def test_require_permission_invalid_token_raises_401(monkeypatch): def test_require_permission_missing_permission_raises_403(monkeypatch): + """_require_permission raises HTTP 403 when the validated user lacks the required + manage_licenses permission.""" monkeypatch.setattr( license_server, "_validate_token", lambda token: {"user_id": 2, "email": "u@omnibioai.test", "permissions": ["manage_config"]}, @@ -96,6 +114,8 @@ def test_require_permission_missing_permission_raises_403(monkeypatch): def test_require_permission_grants_and_returns_user(monkeypatch): + """_require_permission returns the user dict when the validated user holds + manage_licenses.""" monkeypatch.setattr( license_server, "_validate_token", lambda token: {"user_id": 1, "email": "admin@omnibioai.test", "permissions": ["manage_licenses"]}, @@ -108,11 +128,14 @@ def test_require_permission_grants_and_returns_user(monkeypatch): def test_generate_without_token_returns_401(client): + """POST /api/license/generate without a token returns 401.""" resp = client.post("/api/license/generate", json={"email": "x@example.com"}) assert resp.status_code == 401 def test_generate_without_manage_licenses_permission_returns_403(client, monkeypatch): + """POST /api/license/generate with a valid token whose user has no permissions + returns 403.""" monkeypatch.setattr( license_server, "_validate_token", lambda token: {"user_id": 3, "email": "nope@omnibioai.test", "permissions": []}, @@ -125,6 +148,8 @@ def test_generate_without_manage_licenses_permission_returns_403(client, monkeyp def test_generate_with_manage_licenses_permission_succeeds(client, monkeypatch): + """POST /api/license/generate with the manage_licenses permission returns 200 with + the requested email and a key; the database connection is mocked.""" monkeypatch.setattr( license_server, "_validate_token", lambda token: {"user_id": 1, "email": "admin@omnibioai.test", "permissions": ["manage_licenses"]}, @@ -150,6 +175,7 @@ def test_generate_request_body_no_longer_accepts_admin_key_field(): def test_list_licenses_without_token_returns_401(client): + """GET /api/license/list without a token returns 401.""" resp = client.get("/api/license/list") assert resp.status_code == 401 @@ -163,6 +189,8 @@ def test_list_licenses_no_longer_accepts_admin_key_query_param(client, monkeypat def test_list_licenses_with_manage_licenses_permission_succeeds(client, monkeypatch): + """GET /api/license/list with the manage_licenses permission returns 200 and an + empty list when the mocked database has no rows.""" monkeypatch.setattr( license_server, "_validate_token", lambda token: {"user_id": 1, "email": "admin@omnibioai.test", "permissions": ["manage_licenses"]}, @@ -183,6 +211,8 @@ def test_list_licenses_with_manage_licenses_permission_succeeds(client, monkeypa def test_validate_endpoint_unaffected_by_iam_changes(client, monkeypatch): + """POST /api/license/validate stays unauthenticated: with no Authorization header it + returns valid=True for a mocked valid license.""" monkeypatch.setattr( license_server, "_load_valid_license", lambda key: { @@ -199,12 +229,16 @@ def test_validate_endpoint_unaffected_by_iam_changes(client, monkeypatch): def test_verify_service_identity_skips_when_not_configured(monkeypatch): + """verify_service_identity returns False without contacting the auth service when no + service client id or secret is configured.""" monkeypatch.setattr(license_server, "STUDIO_SERVICE_CLIENT_ID", "") monkeypatch.setattr(license_server, "STUDIO_SERVICE_CLIENT_SECRET", "") assert license_server.verify_service_identity() is False def test_verify_service_identity_succeeds_end_to_end(monkeypatch): + """verify_service_identity returns True after a client_credentials token request to + /oauth/token and a bearer-authenticated /service/me call, both mocked.""" monkeypatch.setattr(license_server, "STUDIO_SERVICE_CLIENT_ID", "omni_client_test") monkeypatch.setattr(license_server, "STUDIO_SERVICE_CLIENT_SECRET", "shh") @@ -227,6 +261,8 @@ def _get(url, **kwargs): def test_verify_service_identity_fails_gracefully_on_bad_token_response(monkeypatch): + """verify_service_identity returns False when the mocked token endpoint responds + 401.""" monkeypatch.setattr(license_server, "STUDIO_SERVICE_CLIENT_ID", "omni_client_test") monkeypatch.setattr(license_server, "STUDIO_SERVICE_CLIENT_SECRET", "wrong") monkeypatch.setattr(license_server.requests, "post", lambda *a, **kw: _mock_response(401, {})) @@ -235,6 +271,8 @@ def test_verify_service_identity_fails_gracefully_on_bad_token_response(monkeypa def test_verify_service_identity_never_raises_on_network_error(monkeypatch): + """verify_service_identity returns False rather than raising when the auth service + is unreachable.""" monkeypatch.setattr(license_server, "STUDIO_SERVICE_CLIENT_ID", "omni_client_test") monkeypatch.setattr(license_server, "STUDIO_SERVICE_CLIENT_SECRET", "shh")