import json import sys import tempfile import unittest from pathlib import Path FACTORY_HOST_DIR = Path(__file__).resolve().parents[1] sys.path.insert(0, str(FACTORY_HOST_DIR / "src")) import factory_host from ft1_protocol import FT1Session, ProtocolError, parse_ft1_line class FactoryHostTests(unittest.TestCase): def test_expand_args(self): args = factory_host.expand_args( ["--port", "{port}", "{firmware}", "--station", "{station}"], aboot=Path(r"D:\burn_tool\aboot.exe"), firmware=Path(r"D:\fw\prod.zip"), port="COM14", station="2", baudrate=115200, ) self.assertEqual(args[1], "COM14") self.assertEqual(args[2], r"D:\fw\prod.zip") self.assertEqual(args[-1], "2") def test_merge_and_config_round_trip(self): with tempfile.TemporaryDirectory() as temp: path = Path(temp) / "config.json" config = factory_host.load_config(path) config["flash"]["workers"] = 4 factory_host.save_config(config, path) loaded = factory_host.load_config(path) self.assertEqual(loaded["flash"]["workers"], 4) self.assertIn("protocol", loaded) def test_parse_command_template(self): self.assertEqual(factory_host.parse_command_template('"C:\\Program Files\\aboot.exe" --port COM14'), [r"C:\Program Files\aboot.exe", "--port", "COM14"]) def test_default_flash_waits_for_arom_usb(self): args = factory_host.DEFAULT_CONFIG["flash"]["args"] self.assertIn("-u", args) self.assertIn("-a", args) self.assertNotIn("{port}", args) def test_auto_flash_starts_one_usb_listener_before_power_on(self): args = factory_host.DEFAULT_CONFIG["flash"]["auto_args"] self.assertIn("-u", args) self.assertIn("-a", args) self.assertNotIn("-q", args) self.assertNotIn("-m", args) self.assertNotIn("{port}", args) def test_auto_flash_classifies_persistent_downloader_results(self): self.assertEqual( factory_host.classify_downloader_line( " all finished. total time: 23.500s"), "烧录成功", ) self.assertEqual( factory_host.classify_downloader_line("failed to parse configuration"), "失败", ) self.assertIsNone(factory_host.classify_downloader_line( " OFFLINE device open failed num = 0, device list size = 0")) def test_auto_flash_assigns_untagged_result_to_unfinished_port(self): activity = {"COM38": 8, "COM8": 9} self.assertEqual( factory_host.resolve_downloader_result_port( "", "COM8", activity, {"COM8"}), "COM38", ) self.assertEqual( factory_host.resolve_downloader_result_port( "COM38", "COM8", activity, {"COM8"}), "COM38", ) self.assertEqual( factory_host.resolve_downloader_result_port( "COM8", "COM8", activity, {"COM8"}), "", ) self.assertEqual( factory_host.resolve_downloader_result_port( "", "COM8", activity, {"COM8", "COM38"}), "", ) def test_auto_flash_accepts_only_asr_arom_port(self): arom = factory_host.SerialPortInfo( device="COM8", hwid=r"USB\VID_2ECC&PID_3004\1234", vid=0x2ECC, pid=0x3004) arom_from_hwid = factory_host.SerialPortInfo( device="COM22", hwid=r"USB\VID_2ECC&PID_3004\5678") runtime = factory_host.SerialPortInfo( device="COM13", hwid=r"USB\VID_2ECC&PID_3017&MI_00\1234", vid=0x2ECC, pid=0x3017, description="ASR Serial Download Device") unrelated = factory_host.SerialPortInfo( device="COM3", hwid=r"USB\VID_1A86&PID_7523\5678", vid=0x1A86, pid=0x7523) misleading = factory_host.SerialPortInfo( device="COM4", description="Generic BootROM Serial Device") self.assertTrue(factory_host.is_arom_download_port(arom)) self.assertTrue(factory_host.is_arom_download_port(arom_from_hwid)) self.assertFalse(factory_host.is_arom_download_port(runtime)) self.assertFalse(factory_host.is_arom_download_port(unrelated)) self.assertFalse(factory_host.is_arom_download_port(misleading)) def test_auto_flash_ignores_eg800_runtime_interfaces(self): roles = ( ("COM13", 0x03, "Quectel USB AT Port"), ("COM14", 0x05, "USB Serial Device"), ("COM15", 0x04, "Quectel USB Modem"), ("COM16", 0x02, "Quectel USB DIAG Port"), ) for device, interface, description in roles: info = factory_host.SerialPortInfo( device=device, description=description, hwid=rf"USB\VID_2C7C&PID_6002&MI_{interface:02X}\1234", vid=0x2C7C, pid=0x6002, ) with self.subTest(device=device): self.assertFalse(factory_host.is_arom_download_port(info)) def test_auto_flash_groups_interfaces_by_physical_location(self): first = factory_host.SerialPortInfo( device="COM8", vid=0x2ECC, pid=0x3004, location="1-4:1.0") second = factory_host.SerialPortInfo( device="COM9", vid=0x2ECC, pid=0x3004, location="1-4:1.1") ports = factory_host.arom_download_ports([second, first]) self.assertEqual(len(ports), 1) self.assertEqual(next(iter(ports.values())).device, "COM8") def test_auto_flash_port_identity_survives_com_number_change(self): before = factory_host.SerialPortInfo( device="COM8", vid=0x2ECC, pid=0x3004, location="Port_#0004.Hub_#0001") after = factory_host.SerialPortInfo( device="COM21", vid=0x2ECC, pid=0x3004, location="Port_#0004.Hub_#0001") self.assertEqual( factory_host.download_device_key(before), factory_host.download_device_key(after), ) def test_project_root_and_relative_path(self): self.assertTrue((factory_host.FACTORY_HOST_DIR / "vendor" / "aboot-tools-2024.08.08-win-x64").is_dir()) self.assertTrue((factory_host.FACTORY_HOST_DIR / "firmware").is_dir()) self.assertEqual(factory_host.FACTORY_HOST_DIR, FACTORY_HOST_DIR) self.assertEqual(factory_host.CONFIG_PATH, FACTORY_HOST_DIR / "config" / "factory_host_config.json") self.assertEqual(factory_host.LOG_DIR, FACTORY_HOST_DIR / "build" / "logs") config = factory_host.load_config(factory_host.CONFIG_PATH) self.assertEqual( Path(config["firmwares"]["production"]).name, "EG800AKCN_11LG-FT1-factory-test.zip", ) self.assertTrue( factory_host.resolve_project_path(config["firmwares"]["production"]).is_file() ) self.assertTrue( factory_host.resolve_project_path(config["firmwares"]["official"]).is_file() ) self.assertTrue(config["firmwares"]["production"].startswith("firmware/")) self.assertTrue(config["firmwares"]["official"].startswith("firmware/")) self.assertTrue(config["flash"]["aboot_path"].startswith(".runtime/")) self.assertTrue(config["flash"]["downloader_path"].startswith(".runtime/")) def test_ft1_ignores_sdk_log_and_parses_node(self): self.assertIsNone(parse_ft1_line("DP1312EA is OK!")) message = parse_ft1_line( 'FT1 {"v":2,"type":"node_result","seq":3,"session":"BOOT-1",' '"uptime_ms":2450,"imei":"860123456789012","iccid":"","nfc_uid":"",' '"node":1,"name":"POWER_ADC","result":"PASS","code":0,"elapsed_ms":350,' '"data":{"voltage_mv":3865}}' ) self.assertEqual(message["data"]["voltage_mv"], 3865) def test_ft1_session_result_and_sequence_guard(self): session = FT1Session() start = parse_ft1_line( 'FT1 {"v":2,"type":"session_start","seq":1,"session":"BOOT-1",' '"uptime_ms":2050,"imei":"860123456789012","iccid":"","nfc_uid":"",' '"fw":"FACTORY_TEST_1.1.1","start_time":null}' ) session.apply(start) result = parse_ft1_line( 'FT1 {"v":2,"type":"session_result","seq":2,"session":"BOOT-1",' '"uptime_ms":40120,"result":"FAIL","fail_code":1203,"imei":"860123456789012",' '"iccid":"","nfc_uid":"","start_time":null,"duration_ms":38070,"nodes":' '{"BOOT_AUDIO":"PASS","POWER_ADC":"PASS","AUDIO_LOOPBACK":"FAIL",' '"ESIM":"WAIT","DATA_CALL":"WAIT","NFC":"WAIT","GNSS":"WAIT"}}' ) session.apply(result) self.assertEqual(session.result, "FAIL") self.assertEqual(session.nodes["AUDIO_LOOPBACK"], "FAIL") self.assertTrue(session.complete) with self.assertRaises(ProtocolError): session.apply(result) def test_ft1_rejects_node_name_mismatch(self): with self.assertRaises(ProtocolError): parse_ft1_line( 'FT1 {"v":2,"type":"node_start","seq":2,"session":"BOOT-1",' '"uptime_ms":2100,"node":1,"name":"NFC"}' ) def test_ft1_control_messages(self): ready = parse_ft1_line( 'FT1 {"v":2,"type":"device_ready","seq":1,"session":"BOOT-1",' '"uptime_ms":100,"fw":"FACTORY_TEST_1.1.1"}' ) ack = parse_ft1_line( 'FT1 {"v":2,"type":"command_ack","seq":2,"session":"BOOT-1",' '"uptime_ms":120,"command":"START","accepted":true}' ) session = FT1Session() self.assertEqual(session.apply(ready), "设备已识别 FACTORY_TEST_1.1.1") self.assertEqual(session.apply(ack), "START 已确认") self.assertTrue(session.device_ready_seen) self.assertTrue(session.start_ack_seen) def test_ft1_status_snapshot_keeps_passed_nodes(self): first = parse_ft1_line( 'FT1 {"v":2,"type":"status_snapshot","seq":1,"session":"BOOT-1",' '"uptime_ms":100,"active_node":0,"nodes":{"BOOT_AUDIO":"WAIT",' '"POWER_ADC":"WAIT","AUDIO_LOOPBACK":"WAIT","ESIM":"WAIT",' '"DATA_CALL":"WAIT","NFC":"WAIT","GNSS":"WAIT"}}' ) second = parse_ft1_line( 'FT1 {"v":2,"type":"status_snapshot","seq":2,"session":"BOOT-1",' '"uptime_ms":600,"active_node":1,"nodes":{"BOOT_AUDIO":"PASS",' '"POWER_ADC":"WAIT","AUDIO_LOOPBACK":"WAIT","ESIM":"WAIT",' '"DATA_CALL":"WAIT","NFC":"WAIT","GNSS":"WAIT"}}' ) session = FT1Session() session.apply(first) session.apply(second) self.assertEqual(session.nodes["BOOT_AUDIO"], "PASS") self.assertEqual(session.current_node, "POWER_ADC") def test_ft1_parses_low_threshold_gnss_result(self): message = parse_ft1_line( 'FT1 {"v":2,"type":"node_result","seq":9,"session":"BOOT-1",' '"uptime_ms":30600,"imei":"860123456789012","iccid":"8986000000000000000",' '"nfc_uid":"04A1B2C3D4E5F6","node":6,"name":"GNSS","result":"PASS",' '"code":0,"elapsed_ms":800,"data":{"open_ret":0,"nmea_count":5,' '"sat_count":0,"fix_status":0,"search_ms":28500,' '"required_nmea_count":5,"required_sat_count":0}}' ) session = FT1Session() session.apply(message) self.assertEqual(session.nodes["GNSS"], "PASS") self.assertEqual(session.node_details["GNSS"]["data"]["sat_count"], 0) if __name__ == "__main__": unittest.main()