More tests

tests/unit/test_retrieval/test_structured_diag/test_message_translation.py

@@ -0,0 +1,172 @@
+"""
+Unit tests for message translation in structured-diag service
+"""
+
+import pytest
+from trustgraph.messaging.translators.diagnosis import (
+    StructuredDataDiagnosisRequestTranslator,
+    StructuredDataDiagnosisResponseTranslator
+)
+from trustgraph.schema.services.diagnosis import (
+    StructuredDataDiagnosisRequest,
+    StructuredDataDiagnosisResponse
+)
+
+
+class TestRequestTranslation:
+    """Test request message translation"""
+
+    def test_translate_schema_selection_request(self):
+        """Test translating schema-selection request from API to Pulsar"""
+        translator = StructuredDataDiagnosisRequestTranslator()
+
+        # API format (with hyphens)
+        api_data = {
+            "operation": "schema-selection",
+            "sample": "test data sample",
+            "options": {"filter": "catalog"}
+        }
+
+        # Translate to Pulsar
+        pulsar_msg = translator.to_pulsar(api_data)
+
+        assert pulsar_msg.operation == "schema-selection"
+        assert pulsar_msg.sample == "test data sample"
+        assert pulsar_msg.options == {"filter": "catalog"}
+
+    def test_translate_request_with_all_fields(self):
+        """Test translating request with all fields"""
+        translator = StructuredDataDiagnosisRequestTranslator()
+
+        api_data = {
+            "operation": "generate-descriptor",
+            "sample": "csv data",
+            "type": "csv",
+            "schema-name": "products",
+            "options": {"delimiter": ","}
+        }
+
+        pulsar_msg = translator.to_pulsar(api_data)
+
+        assert pulsar_msg.operation == "generate-descriptor"
+        assert pulsar_msg.sample == "csv data"
+        assert pulsar_msg.type == "csv"
+        assert pulsar_msg.schema_name == "products"
+        assert pulsar_msg.options == {"delimiter": ","}
+
+
+class TestResponseTranslation:
+    """Test response message translation"""
+
+    def test_translate_schema_selection_response(self):
+        """Test translating schema-selection response from Pulsar to API"""
+        translator = StructuredDataDiagnosisResponseTranslator()
+
+        # Create Pulsar response with schema_matches
+        pulsar_response = StructuredDataDiagnosisResponse(
+            operation="schema-selection",
+            schema_matches=["products", "inventory", "catalog"],
+            error=None
+        )
+
+        # Translate to API format
+        api_data = translator.from_pulsar(pulsar_response)
+
+        assert api_data["operation"] == "schema-selection"
+        assert api_data["schema-matches"] == ["products", "inventory", "catalog"]
+        assert "error" not in api_data  # None errors shouldn't be included
+
+    def test_translate_empty_schema_matches(self):
+        """Test translating response with empty schema_matches"""
+        translator = StructuredDataDiagnosisResponseTranslator()
+
+        pulsar_response = StructuredDataDiagnosisResponse(
+            operation="schema-selection",
+            schema_matches=[],
+            error=None
+        )
+
+        api_data = translator.from_pulsar(pulsar_response)
+
+        assert api_data["operation"] == "schema-selection"
+        assert api_data["schema-matches"] == []
+
+    def test_translate_response_without_schema_matches(self):
+        """Test translating response without schema_matches field"""
+        translator = StructuredDataDiagnosisResponseTranslator()
+
+        # Old-style response without schema_matches
+        pulsar_response = StructuredDataDiagnosisResponse(
+            operation="detect-type",
+            detected_type="xml",
+            confidence=0.9,
+            error=None
+        )
+
+        api_data = translator.from_pulsar(pulsar_response)
+
+        assert api_data["operation"] == "detect-type"
+        assert api_data["detected-type"] == "xml"
+        assert api_data["confidence"] == 0.9
+        assert "schema-matches" not in api_data  # None values shouldn't be included
+
+    def test_translate_response_with_error(self):
+        """Test translating response with error"""
+        translator = StructuredDataDiagnosisResponseTranslator()
+        from trustgraph.schema.core.primitives import Error
+
+        pulsar_response = StructuredDataDiagnosisResponse(
+            operation="schema-selection",
+            error=Error(
+                type="PromptServiceError",
+                message="Service unavailable"
+            )
+        )
+
+        api_data = translator.from_pulsar(pulsar_response)
+
+        assert api_data["operation"] == "schema-selection"
+        # Error objects are typically handled separately by the gateway
+        # but the translator shouldn't break on them
+
+    def test_translate_all_response_fields(self):
+        """Test translating response with all possible fields"""
+        translator = StructuredDataDiagnosisResponseTranslator()
+        import json
+
+        descriptor_data = {"mapping": {"field1": "column1"}}
+
+        pulsar_response = StructuredDataDiagnosisResponse(
+            operation="diagnose",
+            detected_type="csv",
+            confidence=0.95,
+            descriptor=json.dumps(descriptor_data),
+            metadata={"field_count": "5"},
+            schema_matches=["schema1", "schema2"],
+            error=None
+        )
+
+        api_data = translator.from_pulsar(pulsar_response)
+
+        assert api_data["operation"] == "diagnose"
+        assert api_data["detected-type"] == "csv"
+        assert api_data["confidence"] == 0.95
+        assert api_data["descriptor"] == descriptor_data  # Should be parsed from JSON
+        assert api_data["metadata"] == {"field_count": "5"}
+        assert api_data["schema-matches"] == ["schema1", "schema2"]
+
+    def test_response_completion_flag(self):
+        """Test that response includes completion flag"""
+        translator = StructuredDataDiagnosisResponseTranslator()
+
+        pulsar_response = StructuredDataDiagnosisResponse(
+            operation="schema-selection",
+            schema_matches=["products"],
+            error=None
+        )
+
+        api_data, is_final = translator.from_response_with_completion(pulsar_response)
+
+        assert is_final is True  # Structured-diag responses are always final
+        assert api_data["operation"] == "schema-selection"
+        assert api_data["schema-matches"] == ["products"]

tests/unit/test_retrieval/test_structured_diag/test_type_detection.py

@@ -0,0 +1,179 @@
+"""
+Unit tests for simplified type detection in structured-diag service
+"""
+
+import pytest
+from trustgraph.retrieval.structured_diag.type_detector import detect_data_type
+
+
+class TestSimplifiedTypeDetection:
+    """Test the simplified type detection logic"""
+
+    def test_xml_detection_with_declaration(self):
+        """Test XML detection with XML declaration"""
+        sample = '<?xml version="1.0"?><root><item>data</item></root>'
+        data_type, confidence = detect_data_type(sample)
+        assert data_type == "xml"
+        assert confidence == 0.9
+
+    def test_xml_detection_without_declaration(self):
+        """Test XML detection without declaration but with closing tags"""
+        sample = '<root><item>data</item></root>'
+        data_type, confidence = detect_data_type(sample)
+        assert data_type == "xml"
+        assert confidence == 0.9
+
+    def test_xml_detection_truncated(self):
+        """Test XML detection with truncated XML (common with 500-byte samples)"""
+        sample = '''<?xml version="1.0" encoding="UTF-8"?>
+<pieDataset>
+  <pies>
+    <pie id="1">
+      <pieType>Steak &amp; Kidney</pieType>
+      <region>Yorkshire</region>
+      <diameterCm>12.5</diameterCm>
+      <heightCm>4.2'''  # Truncated mid-element
+        data_type, confidence = detect_data_type(sample)
+        assert data_type == "xml"
+        assert confidence == 0.9
+
+    def test_json_object_detection(self):
+        """Test JSON object detection"""
+        sample = '{"name": "John", "age": 30, "city": "New York"}'
+        data_type, confidence = detect_data_type(sample)
+        assert data_type == "json"
+        assert confidence == 0.9
+
+    def test_json_array_detection(self):
+        """Test JSON array detection"""
+        sample = '[{"id": 1}, {"id": 2}, {"id": 3}]'
+        data_type, confidence = detect_data_type(sample)
+        assert data_type == "json"
+        assert confidence == 0.9
+
+    def test_json_truncated(self):
+        """Test JSON detection with truncated JSON"""
+        sample = '{"products": [{"id": 1, "name": "Widget", "price": 19.99}, {"id": 2, "na'
+        data_type, confidence = detect_data_type(sample)
+        assert data_type == "json"
+        assert confidence == 0.9
+
+    def test_csv_detection(self):
+        """Test CSV detection as fallback"""
+        sample = '''name,age,city
+John,30,New York
+Jane,25,Boston
+Bob,35,Chicago'''
+        data_type, confidence = detect_data_type(sample)
+        assert data_type == "csv"
+        assert confidence == 0.8
+
+    def test_csv_detection_single_line(self):
+        """Test CSV detection with single line defaults to CSV"""
+        sample = 'column1,column2,column3'
+        data_type, confidence = detect_data_type(sample)
+        assert data_type == "csv"
+        assert confidence == 0.8
+
+    def test_empty_input(self):
+        """Test empty input handling"""
+        data_type, confidence = detect_data_type("")
+        assert data_type is None
+        assert confidence == 0.0
+
+    def test_whitespace_only(self):
+        """Test whitespace-only input"""
+        data_type, confidence = detect_data_type("   \n  \t  ")
+        assert data_type is None
+        assert confidence == 0.0
+
+    def test_html_not_xml(self):
+        """Test HTML is detected as XML (has closing tags)"""
+        sample = '<html><body><h1>Title</h1></body></html>'
+        data_type, confidence = detect_data_type(sample)
+        assert data_type == "xml"  # HTML is detected as XML
+        assert confidence == 0.9
+
+    def test_malformed_xml_still_detected(self):
+        """Test malformed XML is still detected as XML"""
+        sample = '<root><item>data</item><unclosed>'
+        data_type, confidence = detect_data_type(sample)
+        assert data_type == "xml"
+        assert confidence == 0.9
+
+    def test_json_with_whitespace(self):
+        """Test JSON detection with leading whitespace"""
+        sample = '   \n  {"key": "value"}'
+        data_type, confidence = detect_data_type(sample)
+        assert data_type == "json"
+        assert confidence == 0.9
+
+    def test_priority_xml_over_csv(self):
+        """Test XML takes priority over CSV when both patterns present"""
+        sample = '<?xml version="1.0"?>\n<data>a,b,c</data>'
+        data_type, confidence = detect_data_type(sample)
+        assert data_type == "xml"
+        assert confidence == 0.9
+
+    def test_priority_json_over_csv(self):
+        """Test JSON takes priority over CSV when both patterns present"""
+        sample = '{"data": "a,b,c"}'
+        data_type, confidence = detect_data_type(sample)
+        assert data_type == "json"
+        assert confidence == 0.9
+
+    def test_text_defaults_to_csv(self):
+        """Test plain text defaults to CSV"""
+        sample = 'This is just plain text without any structure'
+        data_type, confidence = detect_data_type(sample)
+        assert data_type == "csv"
+        assert confidence == 0.8
+
+
+class TestRealWorldSamples:
+    """Test with real-world data samples"""
+
+    def test_uk_pies_xml_sample(self):
+        """Test with actual UK pies XML sample (first 500 bytes)"""
+        sample = '''<?xml version="1.0" encoding="UTF-8"?>
+<pieDataset>
+  <pies>
+    <pie id="1">
+      <pieType>Steak &amp; Kidney</pieType>
+      <region>Yorkshire</region>
+      <diameterCm>12.5</diameterCm>
+      <heightCm>4.2</heightCm>
+      <weightGrams>285</weightGrams>
+      <crustType>Shortcrust</crustType>
+      <fillingCategory>Meat</fillingCategory>
+      <price>3.50</price>
+      <currency>GBP</currency>
+      <bakeryType>Traditional</bakeryType>
+    </pie>
+    <pie id="2">
+      <pieType>Chicken &amp; Mushroom</pieType>
+      <region>Lancashire</regio'''  # Cut at 500 chars
+        data_type, confidence = detect_data_type(sample[:500])
+        assert data_type == "xml"
+        assert confidence == 0.9
+
+    def test_product_json_sample(self):
+        """Test with product catalog JSON sample"""
+        sample = '''{"products": [
+  {"id": "PROD001", "name": "Widget", "price": 19.99, "category": "Tools"},
+  {"id": "PROD002", "name": "Gadget", "price": 29.99, "category": "Electronics"},
+  {"id": "PROD003", "name": "Doohickey", "price": 9.99, "category": "Accessories"}
+]}'''
+        data_type, confidence = detect_data_type(sample)
+        assert data_type == "json"
+        assert confidence == 0.9
+
+    def test_customer_csv_sample(self):
+        """Test with customer CSV sample"""
+        sample = '''customer_id,name,email,signup_date,total_orders
+CUST001,John Smith,john@example.com,2023-01-15,5
+CUST002,Jane Doe,jane@example.com,2023-02-20,3
+CUST003,Bob Johnson,bob@example.com,2023-03-10,7'''
+        data_type, confidence = detect_data_type(sample)
+        assert data_type == "csv"
+        assert confidence == 0.8