"""parametrized 模块的测试。 验证每个公开函数的正常路径和关键边界条件。 """ import os import tempfile import pytest from parametrized import ( generate_matching_data, generate_keybreak_data, generate_division_data, generate_zero_byte_file, generate_boundary_values, generate_minimal_records, generate_sorted_records, generate_duplicate_keys, ) # ── generate_matching_data ── class TestMatchingData: def test_matching_data_basic(self): main, sub = generate_matching_data("1:1", 5, 5) assert len(main) == 5 assert len(sub) == 5 def test_matching_data_imbalance(self): main, sub = generate_matching_data("1:N", 1, 100) assert len(main) == 1 assert len(sub) == 100 def test_matching_n_to_one(self): main, sub = generate_matching_data("N:1", 100, 1) assert len(main) == 100 assert len(sub) == 1 def test_matching_zero_records(self): main, sub = generate_matching_data("1:1", 0, 0) assert len(main) == 0 assert len(sub) == 0 def test_matching_all_unmatched(self): main, sub = generate_matching_data("1:1", 5, 5, key_match_ratio=0.0) assert len(main) == 5 assert len(sub) == 5 # 确认没有匹配的 KEY main_keys = {r["KEY"] for r in main} sub_keys = {r["KEY"] for r in sub} assert main_keys.isdisjoint(sub_keys) def test_matching_all_matched(self): main, sub = generate_matching_data("1:1", 5, 5, key_match_ratio=1.0) assert len(main) == 5 assert len(sub) == 5 main_keys = {r["KEY"] for r in main} sub_keys = {r["KEY"] for r in sub} assert main_keys == sub_keys def test_matching_invalid_type(self): with pytest.raises(ValueError, match="matching_type"): generate_matching_data("INVALID", 5, 5) def test_matching_invalid_ratio(self): with pytest.raises(ValueError, match="key_match_ratio"): generate_matching_data("1:1", 5, 5, key_match_ratio=-0.5) def test_matching_negative_count(self): with pytest.raises(ValueError, match="记录数"): generate_matching_data("1:1", -1, 5) # ── generate_keybreak_data ── class TestKeybreakData: def test_keybreak_data_basic(self): data = generate_keybreak_data(3, 2) assert len(data) >= 6 # 检查 KEY 分组正确 keys = {r["KEY"] for r in data} assert len(keys) == 3 # 3 组 def test_keybreak_data_single_group(self): data = generate_keybreak_data(1, 5) assert len(data) == 5 assert all(r["KEY"] == "KEY-A" for r in data) def test_keybreak_data_accumulate(self): data = generate_keybreak_data(2, 2, sum_type="accumulate") assert len(data) == 4 # GROUP 1: FIELD 值 101, 102 assert data[0]["GROUP"] == 1 assert data[0]["FIELD"] == 101 assert data[1]["FIELD"] == 102 # GROUP 2: FIELD 值 201, 202 assert data[2]["GROUP"] == 2 assert data[2]["FIELD"] == 201 assert data[3]["FIELD"] == 202 def test_keybreak_data_aggregate(self): data = generate_keybreak_data(2, 2, sum_type="aggregate") # 每组值相同 assert data[0]["FIELD"] == 100 assert data[1]["FIELD"] == 100 assert data[2]["FIELD"] == 200 assert data[3]["FIELD"] == 200 def test_keybreak_data_mark(self): data = generate_keybreak_data(2, 1, sum_type="mark") assert data[0]["FIELD"] == "MARK-A" assert data[1]["FIELD"] == "MARK-B" def test_keybreak_invalid_group_count(self): with pytest.raises(ValueError, match="group_count"): generate_keybreak_data(0, 2) def test_keybreak_invalid_sum_type(self): with pytest.raises(ValueError, match="sum_type"): generate_keybreak_data(3, 2, sum_type="unknown") # ── generate_division_data ── class TestDivisionData: def test_division_fifty(self): result = generate_division_data(50, 50) assert len(result) == 2 assert len(result[0]) + len(result[1]) == 50 def test_division_one_hundred(self): result = generate_division_data(100, 50) assert len(result) == 1 assert len(result[0]) == 50 def test_division_twenty_five(self): result = generate_division_data(25, 100) assert len(result) == 4 total = sum(len(f) for f in result) assert total == 100 def test_division_single_record(self): result = generate_division_data(100, 1) assert len(result) == 1 assert len(result[0]) == 1 def test_division_invalid_type(self): with pytest.raises(ValueError, match="division_type"): generate_division_data(99, 50) def test_division_negative_count(self): with pytest.raises(ValueError, match="record_count"): generate_division_data(50, 0) # ── generate_zero_byte_file ── class TestZeroByteFile: def test_zero_byte(self): tmpdir = tempfile.mkdtemp() p = os.path.join(tmpdir, "empty.bin") generate_zero_byte_file(p) assert os.path.getsize(p) == 0 os.remove(p) def test_zero_byte_nested_dir(self): tmpdir = tempfile.mkdtemp() p = os.path.join(tmpdir, "sub", "nested", "empty.dat") generate_zero_byte_file(p) assert os.path.getsize(p) == 0 os.remove(p) # ── generate_boundary_values ── class TestBoundaryValues: def test_boundary_signed_numeric(self): result = generate_boundary_values("S9(7)V99") assert result["max"] == 9999999.99 assert result["min"] == -9999999.99 assert result["overflow"] == 100000000.0 assert result["zero"] == 0.0 def test_boundary_unsigned_integer(self): result = generate_boundary_values("9(4)") assert result["max"] == 9999 assert result["min"] == 0 assert result["overflow"] == 100000 assert result["zero"] == 0 def test_boundary_string(self): result = generate_boundary_values("X(10)") assert result["max"] == "X" * 10 assert result["overflow"] == "X" * 11 def test_boundary_signed_integer(self): result = generate_boundary_values("S9(3)") assert result["max"] == 999 assert result["min"] == -999 assert result["zero"] == 0 # ── generate_minimal_records ── class TestMinimalRecords: def test_minimal_empty_fields(self): records = generate_minimal_records([]) assert records == [{}] def test_minimal_with_fields(self): fields = [ {"name": "ID", "type": "numeric"}, {"name": "NAME", "type": "string", "length": 20}, ] records = generate_minimal_records(fields) assert len(records) == 1 assert records[0]["ID"] == 0 assert len(records[0]["NAME"]) == 20 assert records[0]["NAME"] == "A" * 20 def test_minimal_with_defaults(self): fields = [ {"name": "STATUS", "default": "OK"}, ] records = generate_minimal_records(fields) assert records[0]["STATUS"] == "OK" # ── generate_sorted_records ── class TestSortedRecords: def test_sorted_basic(self): records = generate_sorted_records(5) assert len(records) == 5 assert records[0]["KEY"] == "KEY-0000" assert records[4]["KEY"] == "KEY-0004" def test_sorted_single(self): records = generate_sorted_records(1) assert len(records) == 1 assert records[0]["SEQ"] == 1 def test_sorted_invalid_count(self): with pytest.raises(ValueError, match="record_count"): generate_sorted_records(0) def test_sorted_custom_key(self): records = generate_sorted_records(3, key_field="MYKEY") assert "MYKEY" in records[0] assert records[0]["MYKEY"] == "KEY-0000" # ── generate_duplicate_keys ── class TestDuplicateKeys: def test_duplicate_empty(self): result = generate_duplicate_keys([]) assert result == [] def test_duplicate_basic(self): records = [{"KEY": "K001", "DATA": "a", "SEQ": 1}] result = generate_duplicate_keys(records) assert len(result) == 2 assert result[0]["KEY"] == "K001" assert result[1]["KEY"] == "K001" assert result[1]["DATA"] == "a_DUP" def test_duplicate_multiple(self): records = [ {"KEY": "K001", "DATA": "a", "SEQ": 1}, {"KEY": "K002", "DATA": "b", "SEQ": 2}, ] result = generate_duplicate_keys(records) assert len(result) == 4 assert result[2]["KEY"] == "K001" # dup of first assert result[3]["KEY"] == "K002" # dup of second