现有组件功能与Amesim组件比较对齐,补全功能;仿真服务启动环境检测,旧版json工程文件适配读取

This commit is contained in:
ljz committed 2026-09-13 20:01:56 +08:00
1 parent b6c22a54f8
commit 6fc9afe41d
119 files changed
+260114 -351

No files matched your search

+118
View File
@@ -0,0 +1,118 @@
"""Compiler launch recovery must not hide deterministic compiler diagnostics."""
from __future__ import annotations
from contextlib import ExitStack
import os
from pathlib import Path
import subprocess
import tempfile
import unittest
from unittest.mock import patch
from app.simulation.native_codegen import build, processes
class NativeCompilerProcessTests(unittest.TestCase):
def mocks(self, responses, *, windows=True):
stack=ExitStack();self.addCleanup(stack.close)
stack.enter_context(patch.object(build,'WINDOWS',windows))
stack.enter_context(patch.object(processes,'IS_WINDOWS',windows))
stack.enter_context(patch.object(subprocess,'CREATE_NO_WINDOW',0x08000000,create=True))
stack.enter_context(patch.object(build.LOGGER,'warning'))
run=stack.enter_context(patch.object(subprocess,'run',side_effect=responses))
sleep=stack.enter_context(patch.object(build.time,'sleep'))
return run,sleep
@staticmethod
def result(code=0,stdout=b'ok',stderr=b''):
return subprocess.CompletedProcess(['gcc'],code,stdout,stderr)
def test_transient_child_launch_failure_recovers_and_keeps_history(self):
failure=self.result(1,b'',b'gcc.exe: error: CreateProcess: No such file or directory\n')
run,sleep=self.mocks([failure,failure,self.result(stdout=b'preprocessed')])
log=[]
self.assertEqual(build._command(['gcc','-E','model.c'],log=log),b'preprocessed')
self.assertEqual(run.call_count,3);self.assertEqual(sleep.call_count,2)
self.assertEqual(sum(line.startswith('Compiler launch retry ') for line in log),2)
self.assertEqual(run.call_args.kwargs['stdin'],subprocess.DEVNULL)
self.assertTrue(run.call_args.kwargs['close_fds'])
self.assertEqual(run.call_args.kwargs['creationflags'],0x08000000)
def test_c_and_link_errors_are_not_retried(self):
for diagnostic in (b'model.c:12: error: syntax error',b'ld: undefined reference to missing',
b'model.c:1: error: #error gcc.exe: error: CreateProcess: fake'):
with self.subTest(diagnostic=diagnostic):
run,sleep=self.mocks([self.result(1,b'',diagnostic)])
with self.assertRaises(build.NativeCommandError) as caught:
build._command(['gcc','-c','model.c'],log=[])
self.assertEqual(caught.exception.details['stderr'],diagnostic.decode())
self.assertEqual(run.call_count,1);sleep.assert_not_called()
def test_retry_exhaustion_retains_final_command_and_error(self):
failure=self.result(1,b'',b'gcc.exe: error: CreateProcess: No such file or directory')
run,sleep=self.mocks([failure]*6)
with self.assertRaises(build.NativeCommandError) as caught:
build._command(['gcc','-E','model.c'],log=[])
self.assertEqual(run.call_count,6);self.assertEqual(sleep.call_count,5)
self.assertEqual(caught.exception.details['attempts'],6)
self.assertEqual(caught.exception.details['command'],['gcc','-E','model.c'])
self.assertEqual(caught.exception.details['exitCode'],1)
def test_empty_version_output_is_validated_and_can_recover(self):
run,_=self.mocks([self.result(stdout=b''),self.result(stdout=b'gcc 8.1\n')])
result=build._command(['gcc','--version'],log=[],require_stdout=True)
self.assertEqual(result,b'gcc 8.1\n');self.assertEqual(run.call_count,2)
def test_empty_linux_version_has_a_clear_error_without_windows_retries(self):
run,sleep=self.mocks([self.result(stdout=b'')],windows=False)
with self.assertRaises(build.NativeCommandError) as caught:
build._command(['/usr/bin/gcc','--version'],log=[],require_stdout=True)
self.assertEqual(caught.exception.details['errorType'],'EmptyCompilerOutput')
self.assertEqual(run.call_count,1);sleep.assert_not_called()
self.assertIsNone(run.call_args.kwargs['env'])
self.assertEqual(run.call_args.kwargs['creationflags'],0)
def test_timeout_and_missing_executable_are_not_retried(self):
for exc in (subprocess.TimeoutExpired(['gcc'],1),FileNotFoundError('gcc absent')):
run,sleep=self.mocks([exc])
with self.assertRaises(type(exc)):
build._command(['gcc'],log=[])
self.assertEqual(run.call_count,1);sleep.assert_not_called()
def test_windows_environment_is_per_process_and_uses_selected_toolchain(self):
with tempfile.TemporaryDirectory() as directory:
compiler=Path(directory)/'gcc.exe';compiler.touch()
before=dict(os.environ)
with patch.object(processes,'IS_WINDOWS',True):
env=processes.command_environment(str(compiler))
key=next(k for k in env if k.upper()=='PATH')
self.assertTrue(env[key].startswith(str(compiler.parent.resolve())+os.pathsep))
self.assertEqual(dict(os.environ),before)
with patch.object(processes,'IS_WINDOWS',False):
self.assertIsNone(processes.command_environment(str(compiler)))
def test_driver_error_classifier_does_not_assume_missing_file(self):
with patch.object(processes,'IS_WINDOWS',True):
self.assertTrue(processes.is_driver_launch_failure('gcc.EXE: error: CreateProcess: No such file or directory'))
self.assertTrue(processes.is_driver_launch_failure("x86_64-w64-mingw32-gcc.exe: fatal error: cannot execute 'cc1': CreateProcess: Permission denied"))
self.assertFalse(processes.is_driver_launch_failure('source.c:1: error: #error gcc.exe: error: CreateProcess: fake'))
with patch.object(processes,'IS_WINDOWS',False):
self.assertFalse(processes.is_driver_launch_failure('gcc.exe: error: CreateProcess: No such file or directory'))
def test_startup_check_keeps_exhausted_identity_query_details(self):
from threading import Event
from app.simulation.native_codegen import self_test
error=build.NativeCommandError('empty identity', ['gcc','--version'], exit_code=0,
error_type='EmptyCompilerOutput',attempts=6)
state={}
with patch.object(self_test,'toolchain',side_effect=error):
with self.assertRaises(self_test.CheckFailure) as caught:
self_test.check_native_runtime(lambda **values: state.update(values),Event())
self.assertEqual(state['stage'],'toolchain')
self.assertEqual(caught.exception.details['command'],['gcc','--version'])
self.assertEqual(caught.exception.details['attempts'],6)
self.assertEqual(caught.exception.details['errorType'],'EmptyCompilerOutput')
if __name__=='__main__':
unittest.main()