import copy from datetime import datetime, timezone from decimal import Decimal, localcontext import unittest from car_tracker.transform import clean_observation OBSERVED_AT = datetime(2026, 9, 8, tzinfo=timezone.utc) FIELDS = { "price_eur", "currency", "make", "model", "model_year", "mileage_km", "fuel", "gearbox", "body_style", "power_hp", "fiscal_power_cv", "source_published_at", "source_updated_at", "title", "description", "city", "postcode", "department_code", "latitude", "longitude", "location_precision", } class CleanObservationTests(unittest.TestCase): def clean(self, payload): before = copy.deepcopy(payload) result, issues = clean_observation(payload, OBSERVED_AT) self.assertEqual(set(result), FIELDS) self.assertEqual(payload, before) for issue in issues: self.assertEqual(set(issue), {"field", "code"}) self.assertIsInstance(issue["field"], str) self.assertIsInstance(issue["code"], str) return result, issues def assert_issue(self, issues, field, code): self.assertIn({"field": field, "code": code}, issues) def test_missing_values(self): for payload in ({}, {"list_id": 123}, {"price": []}, {"price": None, "price_cents": None, "attributes": None, "location": None, "subject": None, "body": None}): with self.subTest(payload=payload): result, issues = self.clean(payload) self.assertTrue(all(value is None for value in result.values())) self.assertEqual(issues, []) def test_price_units(self): for payload, expected in ( ({"price": [1234]}, "1234.00"), ({"price": ["1234.50"]}, "1234.50"), ({"price_cents": 123450}, "1234.50"), ({"price": [], "price_cents": "1"}, "0.01"), ({"price": [12.5], "price_cents": "1250"}, "12.50"), ({"price": [0], "price_cents": 0}, "0.00"), ({"price": ["999999999999.99"], "price_cents": 99999999999999}, "999999999999.99"), ({"price": [Decimal("12.34000")]}, "12.34"), ): with self.subTest(payload=payload): result, issues = self.clean(payload) self.assertEqual(result["price_eur"], Decimal(expected)) self.assertIsInstance(result["price_eur"], Decimal) self.assertEqual(result["price_eur"].as_tuple().exponent, -2) self.assertEqual(result["currency"], "EUR") self.assertEqual(issues, []) def test_bad_price_does_not_fall_back_to_other_unit(self): cases = [ ({"price": [10], "price_cents": 10}, "conflicting_prices"), ({"price": [10, 10], "price_cents": 1000}, "ambiguous_value"), ({"price": [10], "price_cents": True}, "invalid_value"), ({"price": ["10 euros"], "price_cents": 1000}, "invalid_value"), ({"price": "10", "price_cents": 1000}, "invalid_value"), ({"price": 10}, "invalid_value"), ] for payload, code in cases: with self.subTest(payload=payload): result, issues = self.clean(payload) self.assertIsNone(result["price_eur"]) self.assertIsNone(result["currency"]) self.assert_issue(issues, "price_eur", code) def test_invalid_prices(self): for value in (True, False, float("inf"), float("nan"), Decimal("NaN"), Decimal("-Infinity"), -1, "-0.01", "1000000000000", "1.001", "1.000000000000000000000000000001", "1,50", "1 000", "1.000,00", "100 EUR", "1e3", "", {}, [[1]]): with self.subTest(value=value): result, issues = self.clean({"price": [value]}) self.assertIsNone(result["price_eur"]) self.assertIsNone(result["currency"]) self.assertTrue(issues) for value in (1.5, "1.01", -1, 100000000000000, True, "NaN", []): with self.subTest(cents=value): result, issues = self.clean({"price_cents": value}) self.assertIsNone(result["price_eur"]) self.assertTrue(issues) def test_decimal_context_does_not_round_prices(self): with localcontext() as context: context.prec = 4 result, issues = self.clean({"price_cents": 99999999999999}) self.assertEqual(result["price_eur"], Decimal("999999999999.99")) self.assertEqual(issues, []) result, issues = self.clean({"price": [Decimal("1.000000000000000000000001")]}) self.assertIsNone(result["price_eur"]) self.assert_issue(issues, "price_eur", "invalid_precision") def test_source_attributes_and_labels(self): attributes = [ {"key": "brand", "value": "BMW", "value_label": "Not the source brand"}, {"key": "model", "values": ["Serie 3"], "value_label": "E30 325i"}, {"key": "regdate", "value": "1990"}, {"key": "mileage", "value": "150000"}, {"key": "horse_power_din", "value": "170"}, {"key": "horsepower", "value": "10"}, {"key": "fuel", "value": "1", "value_label": "Essence"}, {"key": "gearbox", "value": "2", "value_label": "Automatique"}, {"key": "vehicle_type", "values": ["3"], "value_label": "Coupe"}, {"key": "search_label", "value": "E30 generation sport variant"}, ] result, issues = self.clean({"attributes": attributes, "brand": "ignored"}) expected = {"make": "BMW", "model": "Serie 3", "model_year": 1990, "mileage_km": 150000, "power_hp": 170, "fiscal_power_cv": 10, "fuel": "Essence", "gearbox": "Automatique", "body_style": "Coupe"} for field, value in expected.items(): self.assertEqual(result[field], value) self.assertIs(type(result[field]), type(value)) self.assertEqual(issues, []) def test_observed_model_taxonomy_and_fiscal_power(self): for values in (["900", "model_all"], ["model_all", "900"]): with self.subTest(values=values): result, issues = self.clean({"brand": "SAAB", "attributes": [ {"key": "model", "value": "900", "values": values}, {"key": "u_car_model", "value": "SAAB_900", "values": ["SAAB_900", "SAAB_model_all"]}, {"key": "horsepower", "value": "8"}, {"key": "horse_power_din", "value": "145"}, ]}) self.assertEqual(result["make"], "SAAB") self.assertEqual(result["model"], "900") self.assertEqual(result["fiscal_power_cv"], 8) self.assertEqual(result["power_hp"], 145) self.assertEqual(issues, []) result, issues = self.clean({"attributes": [ {"key": "horse_power", "value": "8"}, {"key": "u_car_model", "value": "SAAB_900", "values": ["SAAB_900", "SAAB_model_all"]}, ]}) self.assertIsNone(result["fiscal_power_cv"]) self.assertIsNone(result["model"]) self.assertEqual(issues, []) def test_model_taxonomy_does_not_hide_conflicts(self): for value, values in ( ("900", ["900", "9000", "model_all"]), ("900", ["9000", "model_all"]), ("900", ["900", "unrelated_all"]), ("900", ["900", "SAAB_model_all"]), ("900", ["model_all"]), (None, ["900", "model_all"]), ("", ["900", "model_all"]), (" ", ["900", "model_all"]), ("900", ["900", {}, "model_all"]), ): with self.subTest(value=value, values=values): result, issues = self.clean({"attributes": [ {"key": "model", "value": value, "values": values}]}) self.assertIsNone(result["model"]) self.assertTrue(issues) for second, expected in (("900", "900"), ("9000", None)): result, issues = self.clean({"attributes": [ {"key": "model", "value": "900", "values": ["900", "model_all"]}, {"key": "model", "value": second}, ]}) self.assertEqual(result["model"], expected) if expected is None: self.assert_issue(issues, "model", "conflicting_attributes") else: self.assertEqual(issues, []) result, issues = self.clean({"attributes": [ {"key": "brand", "value": "SAAB", "values": ["SAAB", "model_all"]}]}) self.assertIsNone(result["make"]) self.assert_issue(issues, "make", "ambiguous_value") def test_top_level_brand_fallback_only_without_attribute(self): for payload in ({"brand": "SAAB"}, {"brand": "SAAB", "attributes": None}, {"brand": "SAAB", "attributes": []}, {"brand": "SAAB", "attributes": [{"key": "model", "value": "900"}]}): result, issues = self.clean(payload) self.assertEqual(result["make"], "SAAB") self.assertEqual(issues, []) for attributes, expected in ( ([{"key": "brand", "value": "BMW"}], "BMW"), ([{"key": "brand", "value": "BMW"}, {"key": "brand", "value": "Audi"}], None), ([{"key": "brand", "value": None}], None), ([{"key": "brand", "value": ""}], None), ([{"key": "brand", "value": True}], None), ([{"key": "brand", "values": ["BMW", "Audi"]}], None), ({"brand": "BMW"}, None), ): with self.subTest(attributes=attributes): result, issues = self.clean({"brand": "SAAB", "attributes": attributes}) self.assertEqual(result["make"], expected) for value in (True, 123, {}, [], ""): result, issues = self.clean({"brand": value}) self.assertIsNone(result["make"]) self.assert_issue(issues, "make", "invalid_value") def test_unknown_codes_and_text_categories(self): for field, key in (("fuel", "fuel"), ("gearbox", "gearbox"), ("body_style", "vehicle_type")): for code in ("999", 999, "1.0"): with self.subTest(field=field, code=code): result, issues = self.clean({"attributes": [{"key": key, "value": code}]}) self.assertIsNone(result[field]) self.assert_issue(issues, field, "unmapped_code") for attribute in ({"key": key, "value": "Source category"}, {"key": key, "value_label": "Source category"}): result, issues = self.clean({"attributes": [attribute]}) self.assertEqual(result[field], "Source category") self.assertEqual(issues, []) def test_numeric_attributes_bounds(self): for field, key, low, high in ( ("model_year", "regdate", 1886, 2027), ("mileage_km", "mileage", 0, 10_000_000), ("power_hp", "horse_power_din", 1, 3000), ("fiscal_power_cv", "horsepower", 1, 1000), ): for value in (low, str(high), f"{low}.0"): with self.subTest(field=field, value=value): result, issues = self.clean({"attributes": [{"key": key, "value": value}]}) self.assertEqual(result[field], int(Decimal(value))) self.assertIs(type(result[field]), int) self.assertEqual(issues, []) for value in (low - 1, high + 1, True, False, 1.5, "1,000", "100 km", float("inf"), "NaN", [], {}, "", "2026-01-01"): with self.subTest(field=field, value=value): result, issues = self.clean({"attributes": [{"key": key, "value": value}]}) self.assertIsNone(result[field]) self.assertTrue(issues) def test_duplicate_attributes(self): for key, field, first, second in ( ("brand", "make", "BMW", "Audi"), ("model", "model", "Serie 3", "Serie 5"), ("mileage", "mileage_km", "100", "200"), ("fuel", "fuel", "1", "2"), ): with self.subTest(key=key): result, issues = self.clean({"attributes": [ {"key": key, "value": first}, {"key": key, "value": second}]}) self.assertIsNone(result[field]) self.assert_issue(issues, field, "conflicting_attributes") result, issues = self.clean({"attributes": [ {"key": "fuel", "value": "1", "value_label": "Essence"}, {"key": "fuel", "value": "1", "value_label": "Diesel"}]}) self.assertIsNone(result["fuel"]) self.assert_issue(issues, "fuel", "conflicting_attributes") result, issues = self.clean({"attributes": [ {"key": "mileage", "value": "100"}, {"key": "mileage", "values": ["100"]}]}) self.assertEqual(result["mileage_km"], 100) self.assertEqual(issues, []) def test_attribute_representations(self): for attribute, code in ( ({"key": "model", "values": ["a", "b"]}, "ambiguous_value"), ({"key": "model", "value": "a", "values": ["b"]}, "conflicting_attributes"), ({"key": "model", "values": "a"}, "invalid_value"), ({"key": "model", "value": {}}, "invalid_value"), ({"key": "model", "value": True}, "invalid_value"), ({"key": "model", "value": ""}, "invalid_value"), ): with self.subTest(attribute=attribute): result, issues = self.clean({"attributes": [attribute]}) self.assertIsNone(result["model"]) self.assert_issue(issues, "model", code) result, issues = self.clean({"attributes": [{"key": "model", "values": []}]}) self.assertIsNone(result["model"]) self.assertEqual(issues, []) def test_text_is_unmodified_and_not_inferred(self): title = " BMW E30 325i\n\t" description = "First line\r\nContact private@example.test +33 123456789 " result, issues = self.clean({"subject": title, "body": description}) self.assertEqual(result["title"], title) self.assertEqual(result["description"], description) self.assertIsNone(result["make"]) self.assertIsNone(result["model"]) self.assertEqual(issues, []) result, issues = self.clean({"subject": "", "body": ""}) self.assertEqual(result["title"], "") self.assertEqual(result["description"], "") self.assertEqual(issues, []) for value in (True, 123, [description], {"private": description}): result, issues = self.clean({"subject": value, "body": value}) self.assertIsNone(result["title"]) self.assertIsNone(result["description"]) self.assertNotIn(description, repr(issues)) self.assert_issue(issues, "title", "invalid_value") def test_source_timestamps(self): for field, key in (("source_published_at", "first_publication_date"), ("source_updated_at", "index_date")): for value in ("2026-08-01T10:20:30Z", "2026-08-01 10:20:30+02:00", "2026-08-01T10:20:30.123456-04:00"): result, issues = self.clean({key: value}) self.assertEqual(result[field], datetime.fromisoformat(value)) self.assertIsNotNone(result[field].utcoffset()) self.assertEqual(issues, []) for value in ("2026-08-01T10:20:30", "2026-08-01 10:20:30", "2026-08-01"): result, issues = self.clean({key: value}) self.assertIsNone(result[field]) self.assert_issue(issues, field, "timezone_missing") for value in ("", "2026-02-30T12:00:00Z", "yesterday", True, 123, []): result, issues = self.clean({key: value}) self.assertIsNone(result[field]) self.assert_issue(issues, field, "invalid_value") def test_global_coordinates_and_source_precision(self): for lat, lng in (("-21.1151", "55.5364"), ("4.9", "-52.3"), (-90, -180), (90, 180), (0, 0)): result, issues = self.clean({"location": { "lat": lat, "lng": lng, "city": "Source city", "zipcode": "00123", "department_id": "2A", "origin_type": "city", "type": "point", "source": "provider"}}) self.assertEqual(result["latitude"], Decimal(str(lat))) self.assertEqual(result["longitude"], Decimal(str(lng))) self.assertEqual(result["postcode"], "00123") self.assertEqual(result["department_code"], "2A") self.assertEqual(result["city"], "Source city") self.assertEqual(result["location_precision"], "city") self.assertEqual(issues, []) for location, expected in (({"lat": 48, "lng": 2}, None), ({"source": "provider"}, None), ({"type": "approximate"}, "approximate"), ({"origin_type": "department"}, "department")): result, issues = self.clean({"location": location}) self.assertEqual(result["location_precision"], expected) self.assertEqual(issues, []) def test_invalid_and_incomplete_coordinates(self): for key, field, valid, bad in (("lat", "latitude", 48, 91), ("lng", "longitude", 2, 181)): for value in (bad, -bad, True, float("inf"), "NaN", "48 N", {}, []): location = {"lat": 48, "lng": 2, key: value} with self.subTest(location=location): result, issues = self.clean({"location": location}) self.assertIsNone(result["latitude"]) self.assertIsNone(result["longitude"]) self.assertTrue(any(i["field"] == field for i in issues)) for value in (valid, bad): result, issues = self.clean({"location": {key: value}}) self.assertIsNone(result["latitude"]) self.assertIsNone(result["longitude"]) self.assert_issue(issues, "latitude", "incomplete_coordinates") self.assert_issue(issues, "longitude", "incomplete_coordinates") def test_observed_location_source_with_empty_type_fields(self): for source in ("city", "address"): result, issues = self.clean({"location": { "origin_type": "", "type": "", "source": source, }}) self.assertEqual(result["location_precision"], source) self.assertEqual(issues, []) result, issues = self.clean({"location": {"origin_type": "", "type": ""}}) self.assertIsNone(result["location_precision"]) self.assertEqual(issues, []) def test_malformed_containers_and_location_text(self): for value in (True, "private@example.test", 123, {}): result, issues = self.clean({"attributes": value}) self.assert_issue(issues, "attributes", "invalid_value") for value in (True, "private@example.test", 123, []): result, issues = self.clean({"location": value}) self.assert_issue(issues, "location", "invalid_value") result, issues = self.clean({"attributes": [None, True, {}, {"key": []}]}) self.assert_issue(issues, "attributes", "invalid_value") for key, field in (("city", "city"), ("zipcode", "postcode"), ("department_id", "department_code"), ("origin_type", "location_precision")): for value in (True, 123, [], {}, ""): if key == "origin_type" and value == "": continue result, issues = self.clean({"location": {key: value}}) self.assertIsNone(result[field]) self.assert_issue(issues, field, "invalid_value") if __name__ == "__main__": unittest.main()