diff --git a/cmd/main-worker/server_test.go b/cmd/main-worker/server_test.go index 0968d8b..b20941c 100644 --- a/cmd/main-worker/server_test.go +++ b/cmd/main-worker/server_test.go @@ -1290,3 +1290,146 @@ assert.False(t, resp.IsAdmin) assert.Contains(t, resp.Permissions, auth.PermRead) }) } + +// TestHandleAdminWorkers tests the GET /admin/workers endpoint and verifies +// that workers appear as connected after they subscribe. +func TestHandleAdminWorkers(t *testing.T) { + const adminKey = "test-admin-workers-key" + config := createTestConfigWithAdminKey(t, adminKey) + server, err := NewMainWorkerServer(config) + require.NoError(t, err) + + t.Run("returns empty list when no workers have subscribed", func(t *testing.T) { + req := httptest.NewRequest(http.MethodGet, "/admin/workers", nil) + req.Header.Set("Authorization", adminBearer(adminKey)) + w := httptest.NewRecorder() + server.handleAdminWorkers(w, req) + + assert.Equal(t, http.StatusOK, w.Code) + assert.Equal(t, "application/json", w.Header().Get("Content-Type")) + var workers []map[string]interface{} + require.NoError(t, json.NewDecoder(w.Body).Decode(&workers)) + assert.Empty(t, workers) + }) + + t.Run("shows worker as connected after subscribe", func(t *testing.T) { + // Simulate a processing worker subscribing via the gRPC Subscribe RPC. + // The Main Worker must vend a session token and a master key wrapped + // with the worker's public key before the worker can be considered + // connected. + privKey, pubKey, err := crypto.GenerateRSAKeyPair(2048) + require.NoError(t, err) + pubKeyPEM, err := crypto.MarshalPublicKeyToPEM(pubKey) + require.NoError(t, err) + + subReq := &proto.SubscribeRequest{ + WorkerId: "proc-worker-1", + Pubkey: pubKeyPEM, + Tags: map[string]string{"env": "test"}, + } + subResp, err := server.Subscribe(context.Background(), subReq) + require.NoError(t, err) + + // Verify the subscription response carries the required key material: + // a short-lived worker token and the master key wrapped to the worker's + // RSA public key. Without these the worker cannot decrypt entities. + assert.NotEmpty(t, subResp.Token, "subscribe response must contain a worker session token") + assert.NotEmpty(t, subResp.WrappedKey, "subscribe response must contain the wrapped encryption key") + assert.NotEmpty(t, subResp.KeyId, "subscribe response must identify the encryption key") + + // Confirm the wrapped key can actually be unwrapped with the worker's + // private key — proving the Main Worker encrypted it correctly. + unwrapped, err := crypto.UnwrapKey(privKey, subResp.WrappedKey) + require.NoError(t, err) + assert.Equal(t, server.masterKey, unwrapped, + "unwrapped key must match the Main Worker's master key") + + // The admin/workers endpoint must now list the subscribed worker. + req := httptest.NewRequest(http.MethodGet, "/admin/workers", nil) + req.Header.Set("Authorization", adminBearer(adminKey)) + w := httptest.NewRecorder() + server.handleAdminWorkers(w, req) + + assert.Equal(t, http.StatusOK, w.Code) + + var workers []map[string]interface{} + require.NoError(t, json.NewDecoder(w.Body).Decode(&workers)) + require.Len(t, workers, 1, "expected exactly one connected worker") + + worker := workers[0] + assert.Equal(t, "proc-worker-1", worker["worker_id"]) + assert.Equal(t, "Available", worker["status"], + "worker should be shown as Available (connected)") + }) + + t.Run("shows multiple workers as connected after each subscribes", func(t *testing.T) { + freshConfig := createTestConfigWithAdminKey(t, adminKey) + freshServer, err := NewMainWorkerServer(freshConfig) + require.NoError(t, err) + + workerIDs := []string{"proc-worker-a", "proc-worker-b", "proc-worker-c"} + for _, id := range workerIDs { + privKey, pubKey, err := crypto.GenerateRSAKeyPair(2048) + require.NoError(t, err) + pubKeyPEM, err := crypto.MarshalPublicKeyToPEM(pubKey) + require.NoError(t, err) + + subResp, err := freshServer.Subscribe(context.Background(), &proto.SubscribeRequest{ + WorkerId: id, + Pubkey: pubKeyPEM, + }) + require.NoError(t, err) + + // Each worker must receive a token and the wrapped master key so it + // can decrypt entity data. + assert.NotEmpty(t, subResp.Token) + assert.NotEmpty(t, subResp.WrappedKey) + + // Each worker must be able to unwrap the key it received. + _, err = crypto.UnwrapKey(privKey, subResp.WrappedKey) + require.NoError(t, err, "worker %s must be able to unwrap its key", id) + } + + req := httptest.NewRequest(http.MethodGet, "/admin/workers", nil) + req.Header.Set("Authorization", adminBearer(adminKey)) + w := httptest.NewRecorder() + freshServer.handleAdminWorkers(w, req) + + assert.Equal(t, http.StatusOK, w.Code) + var workers []map[string]interface{} + require.NoError(t, json.NewDecoder(w.Body).Decode(&workers)) + assert.Len(t, workers, len(workerIDs), "all subscribed workers should be shown as connected") + for _, wk := range workers { + assert.Equal(t, "Available", wk["status"]) + } + }) + + t.Run("rejects unauthenticated request with 401", func(t *testing.T) { + req := httptest.NewRequest(http.MethodGet, "/admin/workers", nil) + w := httptest.NewRecorder() + server.handleAdminWorkers(w, req) + + assert.Equal(t, http.StatusUnauthorized, w.Code) + }) + + t.Run("rejects non-admin token with 403", func(t *testing.T) { + secret, _, err := server.keyManager.CreateKey("readonly", []auth.Permission{auth.PermRead}, nil) + require.NoError(t, err) + + req := httptest.NewRequest(http.MethodGet, "/admin/workers", nil) + req.Header.Set("Authorization", "Bearer "+secret) + w := httptest.NewRecorder() + server.handleAdminWorkers(w, req) + + assert.Equal(t, http.StatusForbidden, w.Code) + }) + + t.Run("rejects non-GET method with 405", func(t *testing.T) { + req := httptest.NewRequest(http.MethodPost, "/admin/workers", nil) + req.Header.Set("Authorization", adminBearer(adminKey)) + w := httptest.NewRecorder() + server.handleAdminWorkers(w, req) + + assert.Equal(t, http.StatusMethodNotAllowed, w.Code) + }) +} diff --git a/tests/conftest.py b/tests/conftest.py index 4f5fa34..7924b31 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -69,6 +69,28 @@ def _wait_for_port(host, port, timeout=30.0): return False +def _wait_for_worker(url, admin_key, worker_id, timeout=30.0): + """Poll GET url until worker_id appears in the JSON list or timeout expires. + + Uses the admin key directly as a Bearer token so it can be called before + a session token has been obtained. Returns True when the worker is found, + False on timeout. + """ + deadline = time.monotonic() + timeout + headers = {"Authorization": f"Bearer {admin_key}"} + while time.monotonic() < deadline: + try: + r = _requests.get(url, headers=headers, timeout=2) + if r.status_code == 200: + workers = r.json() + if any(w.get("worker_id") == worker_id for w in workers): + return True + except _requests.RequestException: + pass + time.sleep(0.3) + return False + + @pytest.fixture(scope="session") def live_server(tmp_path_factory): # ── External deployment mode ───────────────────────────────────────────── @@ -130,6 +152,7 @@ def live_server(tmp_path_factory): "token": token, "admin_key": ext_key, "log_path": "", + "proc_worker_id": "", } return # nothing to tear down — the container is managed by CI @@ -169,11 +192,13 @@ def live_server(tmp_path_factory): log_fp.close() pytest.fail(f"main-worker did not start in time. See {log_file}") + proc_worker_id = "session-proc-1" + proc_proc = subprocess.Popen( [ "go", "run", "./cmd/proc-worker", f"-main-addr={main_grpc_addr}", - "-worker-id=session-proc-1", + f"-worker-id={proc_worker_id}", f"-grpc-addr={proc_grpc_addr}", f"-shared-fs={db_dir}", ], @@ -189,6 +214,24 @@ def live_server(tmp_path_factory): log_fp.close() pytest.fail(f"proc-worker did not start in time. See {log_file}") + # Wait for the proc-worker to complete its Subscribe handshake and appear + # in the Main Worker registry. The gRPC port being open only means the + # proc-worker's own server started; the Subscribe call to main-worker runs + # in a separate goroutine and may finish slightly later. + if not _wait_for_worker( + rest_url + "/admin/workers", + "test-admin-key", + proc_worker_id, + timeout=30, + ): + proc_proc.terminate() + main_proc.terminate() + log_fp.close() + pytest.fail( + f"proc-worker '{proc_worker_id}' did not appear in the Main Worker " + f"registry within 30 s. See {log_file}" + ) + try: r = _requests.post( rest_url + "/api/login", @@ -211,6 +254,7 @@ def live_server(tmp_path_factory): "token": token, "admin_key": "test-admin-key", "log_path": str(log_file), + "proc_worker_id": proc_worker_id, } proc_proc.terminate() @@ -236,6 +280,9 @@ def settings(pytestconfig, live_server): "token": live_server["token"], # admin_key can be used directly as a Bearer token — no /api/login needed. "admin_key": live_server["admin_key"], + # proc_worker_id is the worker ID of the proc-worker started by the + # live_server fixture (local mode). Empty string in external mode. + "proc_worker_id": live_server.get("proc_worker_id", ""), } diff --git a/tests/test_apis.py b/tests/test_apis.py index 9e2dc9a..ca9bd2e 100644 --- a/tests/test_apis.py +++ b/tests/test_apis.py @@ -91,6 +91,48 @@ def test_admin_workers_endpoint(settings): assert isinstance(data, list) +def test_admin_workers_shows_connected_worker(settings): + """After the application starts, at least one Processing Worker must appear + as Available in GET /admin/workers. + + The conftest live_server fixture starts both the main-worker and a + proc-worker (worker-id=session-proc-1) and waits for the worker to + complete its Subscribe handshake before yielding. This test confirms + that the worker is actually visible through the REST API. + """ + url = _rest_url(settings, "/admin/workers") + # Use the admin key directly as a Bearer token for unambiguous admin access. + response = requests.get( + url, headers=_auth_header(settings["admin_key"]), timeout=5 + ) + assert response.status_code == 200 + + workers = response.json() + assert isinstance(workers, list), "/admin/workers must return a JSON array" + assert len(workers) >= 1, ( + "Expected at least one connected Processing Worker after startup, " + f"but /admin/workers returned an empty list: {workers}" + ) + + available = [w for w in workers if w.get("status") == "Available"] + assert available, ( + f"No worker has status 'Available' after startup. Workers: {workers}" + ) + + # In local mode the conftest exposes the specific worker ID it started. + # Verify that exact worker is present and Available. + worker_id = settings.get("proc_worker_id", "") + if worker_id: + worker_ids = {w.get("worker_id") for w in workers} + assert worker_id in worker_ids, ( + f"Expected proc-worker '{worker_id}' in the registry but got: {worker_ids}" + ) + matching = next(w for w in workers if w.get("worker_id") == worker_id) + assert matching.get("status") == "Available", ( + f"Worker '{worker_id}' is registered but not Available: {matching}" + ) + + def test_admin_workers_requires_auth(settings): """GET /admin/workers without a token must return 401.""" url = _rest_url(settings, "/admin/workers")