Add pytest integration test verifying a worker appears after application startup

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)
+	})
+}

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", ""),
     }
 
 

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")