""" Unit tests for Co-Memory payload validation. """ import pytest from datetime import datetime from services.memory.qdrant.payload_validation import ( validate_payload, PayloadValidationError, create_minimal_payload, ) class TestPayloadValidation: """Tests for payload validation.""" def test_valid_minimal_payload(self): """Test that minimal valid payload passes validation.""" payload = create_minimal_payload( tenant_id="t_daarion", source_id="doc_01HQ8K9X2NPQR3FGJKLM5678", chunk_id="chk_01HQ8K9X3MPQR3FGJKLM9012", chunk_idx=0, fingerprint="sha256:abc123", team_id="team_core", ) # Should not raise result = validate_payload(payload) assert result["schema_version"] == "cm_payload_v1" assert result["tenant_id"] == "t_daarion" def test_valid_full_payload(self): """Test that full payload with all fields passes validation.""" payload = { "schema_version": "cm_payload_v1", "tenant_id": "t_daarion", "team_id": "team_core", "project_id": "proj_helion", "agent_id": "agt_helion", "owner_kind": "agent", "owner_id": "agt_helion", "scope": "docs", "visibility": "confidential", "indexed": True, "source_kind": "document", "source_id": "doc_01HQ8K9X2NPQR3FGJKLM5678", "chunk": { "chunk_id": "chk_01HQ8K9X3MPQR3FGJKLM9012", "chunk_idx": 0, }, "fingerprint": "sha256:abc123def456", "created_at": "2026-01-26T12:00:00Z", "acl": { "read_team_ids": ["team_core"], "read_agent_ids": ["agt_nutra"], }, "tags": ["product", "features"], "lang": "uk", "importance": 0.8, "embedding": { "model": "cohere-embed-v3", "dim": 1024, "metric": "cosine", } } result = validate_payload(payload) assert result["agent_id"] == "agt_helion" assert result["scope"] == "docs" def test_missing_required_field(self): """Test that missing required field raises error.""" payload = { "schema_version": "cm_payload_v1", "tenant_id": "t_daarion", # Missing other required fields } with pytest.raises(PayloadValidationError) as exc_info: validate_payload(payload) assert "Missing required field" in str(exc_info.value) def test_invalid_schema_version(self): """Test that invalid schema version raises error.""" payload = create_minimal_payload( tenant_id="t_daarion", source_id="doc_01HQ8K9X2NPQR3FGJKLM5678", chunk_id="chk_01HQ8K9X3MPQR3FGJKLM9012", chunk_idx=0, fingerprint="sha256:abc123", ) payload["schema_version"] = "v2" # Invalid with pytest.raises(PayloadValidationError) as exc_info: validate_payload(payload) assert "schema_version" in str(exc_info.value) def test_invalid_tenant_id_format(self): """Test that invalid tenant_id format raises error.""" payload = create_minimal_payload( tenant_id="invalid-tenant", # Wrong format source_id="doc_01HQ8K9X2NPQR3FGJKLM5678", chunk_id="chk_01HQ8K9X3MPQR3FGJKLM9012", chunk_idx=0, fingerprint="sha256:abc123", ) with pytest.raises(PayloadValidationError) as exc_info: validate_payload(payload) assert "tenant_id" in str(exc_info.value) def test_invalid_agent_id_format(self): """Test that invalid agent_id format raises error.""" payload = create_minimal_payload( tenant_id="t_daarion", source_id="doc_01HQ8K9X2NPQR3FGJKLM5678", chunk_id="chk_01HQ8K9X3MPQR3FGJKLM9012", chunk_idx=0, fingerprint="sha256:abc123", ) payload["agent_id"] = "helion" # Missing prefix with pytest.raises(PayloadValidationError) as exc_info: validate_payload(payload) assert "agent_id" in str(exc_info.value) def test_invalid_scope(self): """Test that invalid scope raises error.""" payload = create_minimal_payload( tenant_id="t_daarion", source_id="doc_01HQ8K9X2NPQR3FGJKLM5678", chunk_id="chk_01HQ8K9X3MPQR3FGJKLM9012", chunk_idx=0, fingerprint="sha256:abc123", scope="invalid_scope", ) with pytest.raises(PayloadValidationError) as exc_info: validate_payload(payload) assert "scope" in str(exc_info.value) def test_invalid_visibility(self): """Test that invalid visibility raises error.""" payload = create_minimal_payload( tenant_id="t_daarion", source_id="doc_01HQ8K9X2NPQR3FGJKLM5678", chunk_id="chk_01HQ8K9X3MPQR3FGJKLM9012", chunk_idx=0, fingerprint="sha256:abc123", visibility="secret", # Invalid ) with pytest.raises(PayloadValidationError) as exc_info: validate_payload(payload) assert "visibility" in str(exc_info.value) def test_invalid_importance_range(self): """Test that importance outside 0-1 raises error.""" payload = create_minimal_payload( tenant_id="t_daarion", source_id="doc_01HQ8K9X2NPQR3FGJKLM5678", chunk_id="chk_01HQ8K9X3MPQR3FGJKLM9012", chunk_idx=0, fingerprint="sha256:abc123", ) payload["importance"] = 1.5 # Invalid with pytest.raises(PayloadValidationError) as exc_info: validate_payload(payload) assert "importance" in str(exc_info.value) def test_invalid_chunk_idx_negative(self): """Test that negative chunk_idx raises error.""" payload = create_minimal_payload( tenant_id="t_daarion", source_id="doc_01HQ8K9X2NPQR3FGJKLM5678", chunk_id="chk_01HQ8K9X3MPQR3FGJKLM9012", chunk_idx=-1, # Invalid fingerprint="sha256:abc123", ) with pytest.raises(PayloadValidationError) as exc_info: validate_payload(payload) assert "chunk_idx" in str(exc_info.value) def test_non_strict_mode(self): """Test that non-strict mode returns payload with errors.""" payload = { "schema_version": "cm_payload_v1", "tenant_id": "invalid", # Invalid format } # Should not raise in non-strict mode result = validate_payload(payload, strict=False) assert result["schema_version"] == "cm_payload_v1" def test_all_valid_scopes(self): """Test that all valid scopes pass validation.""" valid_scopes = ["docs", "messages", "memory", "artifacts", "signals"] for scope in valid_scopes: payload = create_minimal_payload( tenant_id="t_daarion", source_id="doc_01HQ8K9X2NPQR3FGJKLM5678", chunk_id="chk_01HQ8K9X3MPQR3FGJKLM9012", chunk_idx=0, fingerprint="sha256:abc123", scope=scope, ) result = validate_payload(payload) assert result["scope"] == scope def test_all_valid_visibilities(self): """Test that all valid visibilities pass validation.""" valid_visibilities = ["public", "confidential", "private"] for visibility in valid_visibilities: payload = create_minimal_payload( tenant_id="t_daarion", source_id="doc_01HQ8K9X2NPQR3FGJKLM5678", chunk_id="chk_01HQ8K9X3MPQR3FGJKLM9012", chunk_idx=0, fingerprint="sha256:abc123", visibility=visibility, ) result = validate_payload(payload) assert result["visibility"] == visibility class TestCollectionNameMapping: """Tests for legacy collection name to payload mapping.""" def test_parse_docs_collection(self): """Test parsing *_docs collection names.""" from scripts.qdrant_migrate_to_canonical import parse_collection_name result = parse_collection_name("helion_docs") assert result is not None assert result["agent_id"] == "agt_helion" assert result["scope"] == "docs" assert result["tags"] == [] def test_parse_messages_collection(self): """Test parsing *_messages collection names.""" from scripts.qdrant_migrate_to_canonical import parse_collection_name result = parse_collection_name("nutra_messages") assert result is not None assert result["agent_id"] == "agt_nutra" assert result["scope"] == "messages" def test_parse_special_kb_collection(self): """Test parsing special knowledge base collections.""" from scripts.qdrant_migrate_to_canonical import parse_collection_name result = parse_collection_name("druid_legal_kb") assert result is not None assert result["agent_id"] == "agt_druid" assert result["scope"] == "docs" assert "legal_kb" in result["tags"] def test_parse_unknown_collection(self): """Test parsing unknown collection returns None.""" from scripts.qdrant_migrate_to_canonical import parse_collection_name result = parse_collection_name("random_collection_xyz") # Should still try to match generic patterns or return None # Based on implementation, this might match *_xyz or return None if __name__ == "__main__": pytest.main([__file__, "-v"])