test_protocols.py

"""
Unit tests for protocol handlers (M4).

Reference: specs/011-madagascar-analyzer-integration, tasks T091–T095.
"""

import json
import os
import unittest
import tempfile
import threading
import http.client
from copy import deepcopy
from http.server import HTTPServer

from protocols.astm_handler import ASTMHandler
from protocols.hl7_handler import HL7Handler
from protocols.serial_handler import SerialHandler
from protocols.file_handler import FileHandler
from api import MockAPIHandler as SimulateAPIHandler


def _load_template(name: str):
    # Use the production loader so these tests exercise the real template path:
    # transport/fixtures from templates/<name>.json + assay fields derived from the
    # canonical profile it references. (A raw json.load would bypass the profile
    # derivation and test a stub.)
    from server import _load_template as _load_production_template
    return _load_production_template(name)


class TestASTMHandler(unittest.TestCase):
    def test_generate_mindray_bc5380(self):
        t = _load_template("mindray_bc5380")
        msg = ASTMHandler().generate(t, patient_id="P001", sample_id="DEV01264000000000001")
        self.assertIn("H|", msg)
        self.assertIn("P|", msg)
        self.assertIn("O|", msg)
        self.assertIn("R|", msg)
        self.assertIn("L|", msg)
        self.assertIn("P001", msg)
        self.assertIn("DEV01264000000000001", msg)

    def test_generate_rejects_invalid_explicit_sample_id(self):
        t = _load_template("mindray_bc5380")
        with self.assertRaisesRegex(ValueError, "valid SiteYearNum accession"):
            ASTMHandler().generate(t, sample_id="S001-BAD")

    def test_generate_horiba_pentra60(self):
        t = _load_template("horiba_pentra60")
        msg = ASTMHandler().generate(t)
        self.assertIn("H|", msg)
        self.assertIn("PENTRA", msg.upper())


class TestASTMGeneXpert(unittest.TestCase):
    """Tests for GeneXpert ASTM template features (PR #13 review comments)."""

    @classmethod
    def setUpClass(cls):
        cls.template = _load_template("genexpert_astm")
        cls.handler = ASTMHandler()

    def _generate(self, **kwargs):
        return self.handler.generate(self.template, use_seed=True, **kwargs)

    def _generate_with_qc(self, **kwargs):
        # Template sets enable_qc=false for the Madagascar harness demo flow.
        # QC generation code path is still covered here via a per-test override.
        t = deepcopy(self.template)
        t.setdefault("astm_config", {})["enable_qc"] = True
        return self.handler.generate(t, use_seed=True, **kwargs)

    def _segments(self, msg, prefix):
        return [l for l in msg.strip().split("\n") if l.startswith(prefix)]

    # --- H-record ---

    def test_h_record_has_message_id(self):
        msg = self._generate()
        h = self._segments(msg, "H")[0]
        fields = h.split("|")
        # H.3 should be a non-empty message ID (MSG-...)
        self.assertTrue(fields[2].startswith("MSG-"), f"H.3 should be MSG-..., got: {fields[2]}")

    def test_h_record_receiver_id(self):
        msg = self._generate()
        h = self._segments(msg, "H")[0]
        fields = h.split("|")
        # H.10 = Receiver ID = "LIS"
        self.assertEqual(fields[9], "LIS")

    def test_h_record_processing_id(self):
        msg = self._generate()
        h = self._segments(msg, "H")[0]
        fields = h.split("|")
        # H.12 = Processing ID = "P"
        self.assertEqual(fields[11], "P")

    def test_h_record_version_number(self):
        msg = self._generate()
        h = self._segments(msg, "H")[0]
        fields = h.split("|")
        # H.13 = Version = "1394-97" (not "LIS2-A2")
        self.assertEqual(fields[12], "1394-97")

    def test_h_record_analyzer_name(self):
        msg = self._generate()
        h = self._segments(msg, "H")[0]
        self.assertIn("GENEXPERT^GeneXpert^4.6.0", h)

    # --- O-record ---

    def test_o_record_action_code_empty_for_patient(self):
        msg = self._generate()
        o = self._segments(msg, "O")[0]
        fields = o.split("|")
        # O.12 (index 11) should be empty for patient samples
        self.assertEqual(fields[11], "")

    def test_o_record_specimen_descriptor(self):
        msg = self._generate()
        o = self._segments(msg, "O")[0]
        fields = o.split("|")
        # O.16 (index 15) = "ORH"
        self.assertEqual(fields[15], "ORH")

    def test_o_record_report_type(self):
        msg = self._generate()
        o = self._segments(msg, "O")[0]
        fields = o.split("|")
        # O.26 (index 25) = "F"
        self.assertEqual(fields[25], "F")

    def test_o_record_universal_test_id(self):
        msg = self._generate()
        o = self._segments(msg, "O")[0]
        fields = o.split("|")
        # O.5 (index 4) = ^^^<first ordered assay code> (assays come from the profile)
        first_code = self.template["fields"][0]["code"]
        self.assertEqual(fields[4], f"^^^{first_code}")

    def test_o_record_has_26_fields(self):
        msg = self._generate()
        o = self._segments(msg, "O")[0]
        fields = o.split("|")
        self.assertEqual(len(fields), 26)

    def test_default_template_generates_valid_siteyearnum_accession(self):
        msg = self._generate()
        o = self._segments(msg, "O")[0]
        sample_id = o.split("|")[2].split("^")[0]
        self.assertRegex(sample_id, r"^DEV01\d{15}$")
        self.assertEqual(len(sample_id), 20)

    def test_invalid_template_sample_seed_fails_loudly(self):
        invalid = deepcopy(self.template)
        invalid["testSample"]["id"] = "SPECIMEN-GX-001"
        with self.assertRaisesRegex(ValueError, "2-digit lane code"):
            self.handler.generate(invalid, use_seed=True)

    # --- R-record: 7-component test ID ---

    def test_r_record_8_component_test_id(self):
        msg = self._generate()
        r_lines = self._segments(msg, "R")
        # The MTB-RIF R-record carries the 8-component GeneXpert test ID (it declares
        # a cartridge version via fieldOverrides). Find it among the derived assays.
        mtb_rif = [l for l in r_lines if l.split("|")[2].split("^")[3:4] == ["MTB-RIF"]]
        self.assertTrue(mtb_rif, "expected an MTB-RIF R-record")
        r_fields = mtb_rif[0].split("|")
        test_id = r_fields[2]
        components = test_id.split("^")
        # ^^^MTB-RIF^Xpert MTB/RIF^2.1^^ = 8 components
        # (3 empty + code + name + version + analyte(empty) + complementary(empty))
        self.assertEqual(len(components), 8, f"Expected 8 components, got {len(components)}: {components}")
        self.assertEqual(components[3], "MTB-RIF")
        self.assertEqual(components[4], "Xpert MTB/RIF")
        self.assertEqual(components[5], "2.1")
        self.assertEqual(components[6], "")  # analyte name (empty for main result)
        self.assertEqual(components[7], "")  # complementary name (empty for main result)

    # --- R-record: complementary results (8 components) ---

    def test_complementary_result_8_component_test_id(self):
        msg = self._generate()
        r_lines = self._segments(msg, "R")
        # Find the Conc/LOG complementary result (HIV-VL has one)
        comp_lines = [r for r in r_lines if "Conc/LOG" in r]
        self.assertTrue(len(comp_lines) > 0, "Should have a Conc/LOG complementary result")
        r_fields = comp_lines[0].split("|")
        test_id = r_fields[2]
        components = test_id.split("^")
        # ^^^HIV-VL^...^...^^Conc/LOG = 8 components
        self.assertEqual(len(components), 8, f"Expected 8 components, got {len(components)}: {components}")
        self.assertEqual(components[7], "Conc/LOG")

    def test_complementary_result_seed_value(self):
        msg = self._generate()
        r_lines = self._segments(msg, "R")
        comp_lines = [r for r in r_lines if "Conc/LOG" in r]
        self.assertTrue(len(comp_lines) > 0)
        r_fields = comp_lines[0].split("|")
        # Value field (R.4) should contain ^3.1 (seed value)
        self.assertIn("3.1", r_fields[3])

    # --- R-record: qualitative values ---

    def test_qualitative_seed_value(self):
        # DATA test: depends on the canonical GeneXpert profile declaring its assays
        # qualitative (result_type). The mock reads projects/analyzer-profiles from
        # its OE2 parent (monorepo layout) — single source of truth. When that
        # profile predates the result_type enrichment (e.g. a standalone CI checkout
        # of an older OE2 ref), there's nothing qualitative to assert, so skip rather
        # than false-fail on un-merged cross-repo state. The adapter's qualitative
        # LOGIC is gated hermetically by TestProfileAdapterQualitativeLogic below, and
        # the real round-trip value is gated by the integrated (OE2-monorepo) harness E2E.
        first_field = self.template["fields"][0]
        if first_field.get("type") != "QUALITATIVE":
            self.skipTest(
                "GeneXpert profile not enriched with result_type (first assay "
                f"'{first_field.get('code')}' is {first_field.get('type')}); qualitative "
                "value is gated by the integrated monorepo CI against PR-version profiles"
            )
        msg = self._generate()
        r_lines = self._segments(msg, "R")
        # First R-record seed should be "NOT DETECTED" — aligns with Cepheid LIS
        # spec 302-2261 vocabulary.
        r_fields = r_lines[0].split("|")
        self.assertIn("NOT DETECTED", r_fields[3])

    # --- QC message generation ---
    # Template's runtime astm_config.enable_qc is false for the harness demo
    # flow (interleaved QC trips up parts of the demo harness). These tests
    # still cover the QC generation code path by overriding enable_qc=true
    # on a per-test copy of the template.

    def test_qc_message_generated(self):
        msg = self._generate_with_qc()
        h_records = self._segments(msg, "H")
        # Should have 2 H-records: one for patient, one for QC
        self.assertEqual(len(h_records), 2, "Should generate patient + QC messages")

    def test_qc_action_code(self):
        msg = self._generate_with_qc()
        o_records = self._segments(msg, "O")
        # Second O-record (QC) should have Action Code "Q" at O.12
        self.assertTrue(len(o_records) >= 2, "Should have at least 2 O-records")
        qc_o_fields = o_records[1].split("|")
        self.assertEqual(qc_o_fields[11], "Q")

    def test_qc_specimen_id(self):
        msg = self._generate_with_qc()
        o_records = self._segments(msg, "O")
        qc_o_fields = o_records[1].split("|")
        # O.3 should contain QC specimen ID
        self.assertIn("DEV01261000000000999", qc_o_fields[2])

    def test_qc_includes_all_template_fields(self):
        msg = self._generate_with_qc()
        # Split into patient msg and QC msg (second H starts QC)
        lines = msg.strip().split("\n")
        h_indices = [i for i, l in enumerate(lines) if l.startswith("H")]
        self.assertEqual(len(h_indices), 2)
        qc_lines = lines[h_indices[1]:]
        qc_r_lines = [l for l in qc_lines if l.startswith("R")]
        # Template has 4 fields + 1 complementary = 5 R-records minimum
        self.assertGreaterEqual(len(qc_r_lines), 4,
                                f"QC should include all template fields, got {len(qc_r_lines)} R-records")

    def test_enable_qc_disabled_by_default_for_harness(self):
        """Template must ship with enable_qc=false so harness demo flow
        doesn't get interleaved QC messages it can't route."""
        self.assertFalse(
            self.template.get("astm_config", {}).get("enable_qc", False),
            "genexpert_astm.json must keep enable_qc=false for the Madagascar harness demo flow",
        )

    # --- proactive_enq config ---

    def test_proactive_enq_config_present(self):
        """GeneXpert template should have proactive_enq enabled."""
        self.assertTrue(
            self.template.get('astm_config', {}).get('proactive_enq'),
            "GeneXpert template should have proactive_enq: true in astm_config"
        )

    def test_proactive_enq_not_in_other_templates(self):
        """Non-GeneXpert templates should NOT have proactive_enq."""
        mindray = _load_template("mindray_bc5380")
        self.assertFalse(
            mindray.get('astm_config', {}).get('proactive_enq'),
            "Mindray template should not have proactive_enq"
        )

    def test_generate_qc_controls_emits_valid_accession_and_o12_q(self):
        """qc_controls path must emit SiteYearNum on O.3 and O.12=Q (GenericASTM index 11)."""
        t = _load_template("genexpert_astm")
        msg = ASTMHandler().generate_qc(t)
        lines = [ln.strip() for ln in msg.strip().split("\n") if ln.startswith("O|")]
        self.assertTrue(lines, "expected at least one O-record")
        o_fields = lines[0].split("|")
        self.assertGreaterEqual(len(o_fields), 12, "O-record must pad to O.12 for QC action code")
        self.assertEqual(o_fields[11], "Q", "O.12 must be Q for QC (0-based index 11)")
        acc = o_fields[2].split("^")[0]
        self.assertRegex(acc, r"^DEV01\d{15}$")
        self.assertEqual(len(acc), 20)

    # --- use_seed determinism ---

    def test_seed_produces_deterministic_output(self):
        msg1 = self._generate()
        msg2 = self._generate()
        # With use_seed=True, qualitative values should be identical
        r1 = [l for l in msg1.strip().split("\n") if l.startswith("R")]
        r2 = [l for l in msg2.strip().split("\n") if l.startswith("R")]
        # Compare R.4 (value field) for qualitative results
        for line1, line2 in zip(r1, r2):
            val1 = line1.split("|")[3]
            val2 = line2.split("|")[3]
            self.assertEqual(val1, val2, f"Seed values should be deterministic: {val1} != {val2}")


class TestHL7Handler(unittest.TestCase):
    def test_generate_mindray_bc5380(self):
        t = _load_template("mindray_bc5380")
        msg = HL7Handler().generate(t, patient_id="P001", sample_id="DEV01264000000000001")
        self.assertIn("MSH|", msg)
        self.assertIn("ORU^R01", msg)
        self.assertIn("PID|", msg)
        self.assertIn("OBR|", msg)
        self.assertIn("OBX|", msg)
        self.assertIn("MINDRAY", msg)
        self.assertIn("P001", msg)
        self.assertIn("DEV01264000000000001", msg)

    def test_invalid_hl7_template_sample_seed_fails_loudly(self):
        t = _load_template("mindray_bc5380")
        invalid = deepcopy(t)
        invalid["testSample"]["id"] = "PLACER-INVALID"
        with self.assertRaisesRegex(ValueError, "2-digit lane code"):
            HL7Handler().generate(invalid)

    def test_invalid_hl7_sample_override_fails_loudly(self):
        t = _load_template("mindray_bc5380")
        with self.assertRaisesRegex(ValueError, "valid SiteYearNum accession"):
            HL7Handler().generate(t, sample_id="S001")

    def test_hl7_mints_valid_accession_when_test_sample_missing(self):
        """If testSample is absent, default lane 00 must produce a valid SiteYearNum."""
        minimal = {
            "protocol": {"type": "HL7", "version": "2.5.1"},
            "identification": {"hl7_sending_app": "TEST", "hl7_sending_facility": "LAB"},
            "fields": [{"code": "GLU", "name": "Glucose", "type": "NUMERIC", "seedValue": 5.0, "unit": "mmol/L"}],
        }
        msg = HL7Handler().generate(minimal)
        self.assertIn("ORC|", msg)
        self.assertIn("OBR|", msg)
        for line in msg.split("\r"):
            if line.startswith("OBR|"):
                obr = line.split("|")
                filler = obr[3].split("^")[0] if len(obr) > 3 else ""
                self.assertRegex(filler, r"^DEV01\d{15}$")
                self.assertEqual(len(filler), 20)
                break
        else:
            self.fail("expected OBR segment")

    def test_generate_sysmex_xn(self):
        t = _load_template("sysmex_xn")
        msg = HL7Handler().generate(t)
        self.assertIn("MSH|", msg)
        self.assertIn("SYSMEX", msg)

    def test_generate_genexpert_hl7_template(self):
        t = _load_template("genexpert")
        msg = HL7Handler().generate(t)
        self.assertIn("GENEXPERT", msg)
        self.assertRegex(msg, r"DEV01\d{15}")


class TestSerialHandler(unittest.TestCase):
    def test_generate_horiba_pentra60(self):
        t = _load_template("horiba_pentra60")
        msg = SerialHandler().generate(t)
        self.assertIn("H|", msg)
        self.assertIn("PENTRA", msg.upper())


class TestFileHandler(unittest.TestCase):
    def test_generate_quantstudio7(self):
        t = _load_template("quantstudio7")
        csv = FileHandler().generate(t, sample_id="DEV01262000000000001")
        self.assertIn("Sample Name", csv)
        self.assertIn("Target Name", csv)
        self.assertIn("Quantity Mean", csv)
        self.assertIn("DEV01262000000000001", csv)

    def test_generate_hain_fluorocycler(self):
        t = _load_template("hain_fluorocycler")
        csv = FileHandler().generate(t)
        self.assertIn("SampleID", csv)
        self.assertIn("TargetName", csv)
        self.assertIn("CP", csv)
        self.assertRegex(csv, r"DEV01\d{15}")

    def test_invalid_file_sample_override_fails_loudly(self):
        t = _load_template("quantstudio7")
        with self.assertRaisesRegex(ValueError, "valid SiteYearNum accession"):
            FileHandler().generate(t, sample_id="S001")


class TestFileSimulateAPI(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        cls.httpd = HTTPServer(("127.0.0.1", 0), SimulateAPIHandler)
        cls.port = cls.httpd.server_address[1]
        cls.thread = threading.Thread(target=cls.httpd.serve_forever, daemon=True)
        cls.thread.start()

    @classmethod
    def tearDownClass(cls):
        cls.httpd.shutdown()
        cls.httpd.server_close()
        cls.thread.join(timeout=10)

    def test_get_simulate_file_quantstudio7(self):
        conn = http.client.HTTPConnection("127.0.0.1", self.port, timeout=5)
        conn.request("GET", "/simulate/file/quantstudio7")
        resp = conn.getresponse()
        body = json.loads(resp.read().decode("utf-8"))
        conn.close()

        self.assertEqual(resp.status, 200)
        self.assertEqual(body.get("status"), "generated")
        # Fixture-backed templates return metadata; synthetic return content
        if "metadata" in body:
            self.assertIn("results", body["metadata"])
            self.assertGreater(len(body["metadata"]["results"]), 0)
        else:
            self.assertIn("Sample Name", body.get("content", ""))

    # ------------------------------------------------------------------
    # Fixture drift guards: every FILE template whose fixture we own must
    # still parse to non-empty metadata results. Catches mismatches between
    # fixture column headers / delimiter / skipRows and the template's
    # fixture.column_mapping — the exact class of bug that broke the
    # Madagascar harness demo flow in CI.
    # ------------------------------------------------------------------

    def _assert_fixture_parses(self, template_name: str):
        conn = http.client.HTTPConnection("127.0.0.1", self.port, timeout=5)
        conn.request("GET", f"/simulate/file/{template_name}")
        resp = conn.getresponse()
        body = json.loads(resp.read().decode("utf-8"))
        conn.close()

        self.assertEqual(resp.status, 200, f"{template_name}: HTTP {resp.status}")
        self.assertEqual(body.get("status"), "generated")
        self.assertIn("metadata", body, f"{template_name}: no metadata in response")
        results = body["metadata"].get("results") or []
        self.assertGreater(
            len(results), 0,
            f"{template_name}: parse_fixture returned 0 results — fixture/profile drift",
        )
        # Every returned result must carry sampleId + result at minimum
        for r in results:
            self.assertIn("sampleId", r, f"{template_name}: result missing sampleId: {r}")
            self.assertIn("result", r, f"{template_name}: result missing value: {r}")

    def test_fixture_parses_hain_fluorocycler(self):
        self._assert_fixture_parses("hain_fluorocycler")

    def test_fixture_parses_wondfo_finecare(self):
        self._assert_fixture_parses("wondfo_finecare")

    def test_fixture_parses_tecan_f50(self):
        self._assert_fixture_parses("tecan_f50")

    def test_fixture_parses_multiskan_fc(self):
        self._assert_fixture_parses("multiskan_fc")

    def test_post_simulate_file_write_target_dir(self):
        with tempfile.TemporaryDirectory() as tmpdir:
            payload = json.dumps({"target_dir": tmpdir, "filename": "sim.csv"})
            headers = {"Content-Type": "application/json"}
            conn = http.client.HTTPConnection("127.0.0.1", self.port, timeout=5)
            conn.request("POST", "/simulate/file/quantstudio7", body=payload, headers=headers)
            resp = conn.getresponse()
            body = json.loads(resp.read().decode("utf-8"))
            conn.close()

            self.assertEqual(resp.status, 200)
            self.assertEqual(body.get("status"), "completed")
            written_path = body.get("written_path")
            self.assertTrue(written_path)
            self.assertTrue(os.path.exists(written_path))


    def test_post_simulate_file_sanitizes_path_traversal_filename(self):
        """Path traversal in filename is stripped to basename (safe write)."""
        with tempfile.TemporaryDirectory() as tmpdir:
            payload = json.dumps({"target_dir": tmpdir, "filename": "../evil.csv"})
            headers = {"Content-Type": "application/json"}
            conn = http.client.HTTPConnection("127.0.0.1", self.port, timeout=5)
            conn.request("POST", "/simulate/file/quantstudio7", body=payload, headers=headers)
            resp = conn.getresponse()
            body = json.loads(resp.read())
            conn.close()
            # Server strips path components — writes "evil.csv" in target_dir (safe)
            self.assertEqual(resp.status, 200)
            self.assertTrue(os.path.exists(os.path.join(tmpdir, "evil.csv")))
            # Verify no file escaped to parent
            self.assertFalse(os.path.exists(os.path.join(os.path.dirname(tmpdir), "evil.csv")))

    def test_post_simulate_file_rejects_invalid_json_body(self):
        headers = {"Content-Type": "application/json"}
        conn = http.client.HTTPConnection("127.0.0.1", self.port, timeout=5)
        conn.request("POST", "/simulate/file/quantstudio7", body="not json", headers=headers)
        resp = conn.getresponse()
        resp.read()
        conn.close()
        self.assertGreaterEqual(resp.status, 400)
        self.assertLess(resp.status, 500)

    def test_simulate_file_rejects_path_traversal_template(self):
        conn = http.client.HTTPConnection("127.0.0.1", self.port, timeout=5)
        conn.request("GET", "/simulate/file/../../etc/passwd")
        resp = conn.getresponse()
        resp.read()
        conn.close()
        self.assertGreaterEqual(resp.status, 400)
        self.assertLess(resp.status, 500)


class TestProfileAdapterQualitativeLogic(unittest.TestCase):
    """Hermetic gate for the adapter's result_type→seed mapping.

    Pins the LOGIC against a SYNTHETIC profile (built in a temp dir and pointed at
    via ANALYZER_PROFILES_DIR) with no dependency on the canonical analyzer-profiles
    tree. This is the mock-OWNED unit gate that runs in standalone CI regardless of
    which OE2 ref is present; the REAL GeneXpert profile values are gated by the
    integrated (OE2-monorepo) harness E2E. Keeps qualitative coverage even when
    test_qualitative_seed_value skips on an un-enriched profile checkout.
    """

    def setUp(self):
        self._tmp = tempfile.TemporaryDirectory()
        self.addCleanup(self._tmp.cleanup)
        astm_dir = os.path.join(self._tmp.name, "astm")
        os.makedirs(astm_dir)
        with open(os.path.join(astm_dir, "synth.json"), "w") as fh:
            json.dump(
                {
                    "default_test_mappings": [
                        {
                            "test_code": "QUAL1",
                            "loinc": "111-1",
                            "result_type": "qualitative",
                            "values": ["DETECTED", "NOT DETECTED"],
                        },
                        {"test_code": "QUANT1", "loinc": "222-2", "result_type": "quantitative"},
                    ]
                },
                fh,
            )
        self._prev = os.environ.get("ANALYZER_PROFILES_DIR")
        os.environ["ANALYZER_PROFILES_DIR"] = self._tmp.name
        self.addCleanup(self._restore_env)

    def _restore_env(self):
        if self._prev is None:
            os.environ.pop("ANALYZER_PROFILES_DIR", None)
        else:
            os.environ["ANALYZER_PROFILES_DIR"] = self._prev

    def test_qualitative_seeds_negative_vocab_quantitative_seeds_zero(self):
        from profile_adapter import load_profile_backed_template

        merged = load_profile_backed_template(
            "synth", {"profile": "astm/synth", "protocol": {"type": "ASTM"}}
        )
        fields = {f["code"]: f for f in merged["fields"]}
        # qualitative result_type -> QUALITATIVE field seeded to the negative vocab
        self.assertEqual(fields["QUAL1"]["type"], "QUALITATIVE")
        self.assertEqual(fields["QUAL1"]["seedQualitative"], "NOT DETECTED")
        # quantitative -> NUMERIC field, deterministic 0 seed
        self.assertEqual(fields["QUANT1"]["type"], "NUMERIC")
        self.assertEqual(fields["QUANT1"]["seedValue"], 0)


if __name__ == "__main__":
    unittest.main()