"""Tests for document model.""" import config import pytest from src.utils import document from src.utils.parser import parse_spec_file @pytest.mark.parametrize( "input_value, expected_output", [ ("Hello\u0014", "Hello"), ("Hello\u0014World", "HelloWorld"), ("Hello\u0019", "Hello"), (None, None), ], ) def test_cleanup_unicode(input_value, expected_output): """Test cleanup_unicode.""" result = document.cleanup_unicode(input_value) assert result == expected_output @pytest.mark.parametrize( "input_value, expected_output", [ ("2025-12-05T01:00:05Z", True), ("2025-12-05", False), ("2025-12-05T10:30:00.123Z", False), ("12/05/2025 10:30:00 AM", False), ("1764933715000", False), (None, False), ], ) def test_is_expected_datetime_format(input_value, expected_output): """Test is_expected_datetime_format.""" result = document.is_expected_datetime_format(input_value) assert result == expected_output def test_prepare_for_upload_delete(test_release_id, product_document_fixture): """Test prepare_for_upload.""" schema = parse_spec_file("releases") expected = { "_index": config.PRODUCT_INDEX_NAME, "_op_type": "delete", "_id": test_release_id, "doc_as_upsert": True, } doc = { **expected, "doc": product_document_fixture, } result = document.prepare_for_upload(schema, doc) assert result == expected LONG_ISRC_LIST = ["" for _ in range(1024)] @pytest.mark.parametrize( "source, cleaned", [ # 0 date_utc field formatting ( {"sale_start_date": "2018-05-14", "release_date": "2018-05-14"}, {"sale_start_date": "2018-05-14T00:00:00Z", "release_date": "2018-05-14T00:00:00Z"}, ), # 1 date_utc field null ( { "release_id": 55555, "sale_start_date": None, "release_date": None, "preorder_date": None, }, {"release_id": 55555}, ), # 2 date_utc field zero ( { "release_id": 55555, "sale_start_date": "0000-00-00", "release_date": "0000-00-00", "preorder_date": "0000-00-00", }, {"release_id": 55555}, ), # 3 date_utc field year is less than minimal ( { "release_id": 55555, "sale_start_date": "0007-08-15", }, {"release_id": 55555}, ), # 4 date_utc field has stale data ( { "release_id": 1111, "sale_start_date": "abcd-08-15", }, {"release_id": 1111}, ), # 5 integer field is null ( {"release_id": None, "product_type_id": 0, "release_date": "2018-05-14"}, {"product_type_id": 0, "release_date": "2018-05-14T00:00:00Z"}, ), # 6 string value with unicode ( { "release_id": 5555, "release_name": "Hello\u0014", "vendor_catalog_number": "Hello\u0019 Pizza\u001a", "track_list": ["Hello\u0019 Pizza\u001a"], }, { "release_id": 5555, "release_name": "Hello", "vendor_catalog_number": "Hello Pizza", "track_list": ["Hello Pizza"], }, ), # 7 string value empty ( {"release_id": 5555, "product_code": None, "vendor_catalog_number": ""}, {"release_id": 5555, "vendor_catalog_number": ""}, ), # 8 array value empty ( {"isrc_list": None, "track_list": ["bees", "lasers", "pizza"]}, {"isrc_list": [], "track_list": ["bees", "lasers", "pizza"]}, ), # 9 fields translate to {} ({"release_id": None, "release_name": None, "sale_start_date": None}, None), # 10 long ISRC list ({"isrc_list": LONG_ISRC_LIST}, None), # 11 long track_type list ({"track_type": LONG_ISRC_LIST}, None), # 12 not existing field ( {"release_id": 5555, "vendor_catalog_number": "", "non_existing_field": "remove me"}, {"release_id": 5555, "vendor_catalog_number": ""}, ), # 13 not_for_distribution field has AccountingDummy ( {"release_id": 5555, "not_for_distribution": "AccountingDummy"}, {"release_id": 5555, "not_for_distribution": "AccountingDummy"}, ), # 14 not_for_distribution field is blank ( {"release_id": 5555, "not_for_distribution": None}, { "release_id": 5555, }, ), ], ) def test_prepare_for_upload(source, cleaned, test_release_id): """Test unicode cleanup.""" schema = parse_spec_file("releases") base = { "_index": config.PRODUCT_INDEX_NAME, "_op_type": "update", "_id": test_release_id, "doc_as_upsert": True, } doc = { **base, "doc": source, } expected = {**base, "doc": cleaned} if cleaned else None prepared = document.prepare_for_upload(schema, doc) assert expected == prepared