Unstract Connector 测试模式全解：从 Mock 模板到集成测试的完整实践

Unstract Connector 测试模式全解从 Mock 模板到集成测试的完整实践【免费下载链接】unstractLLM-Driven Extraction of Unstructured Data — Built for API Deployments ETL Pipeline Workflows项目地址: https://gitcode.com/GitHub_Trending/un/unstract本文基于 Unstract 仓库内 connector-ops 技能参考文档test_patterns.md系统讲解 Unstract 连接器database / filesystem / queue 三类的测试写法如何组织测试文件、如何编写“永远可运行”的 Mock 测试模板与依赖真实服务的集成测试模板、如何用conftest.py提供共享 fixture以及如何用uv run pytest运行整套测试。读完后你可以为任意一个新连接器如 MongoDB、Redis、S3 类对象存储直接套用这些模板生成可落地的测试并从unstract/connectors的源码实现中理解每个断言背后的契约。测试文件组织方式参考文档首先给出了推荐的测试目录组织方式。测试统一放在unstract/connectors/tests/下按连接器类型databases / filesystems / queues分目录每个目录包含__init__.py、mock 测试文件、以_integration后缀命名的集成测试文件以及可选的共享conftest.pytests/ ├── __init__.py ├── test_connectorkit.py # 注册表registry测试 ├── databases/ │ ├── __init__.py │ ├── test_postgresql.py # Mock 测试 │ ├── test_postgresql_integration.py # 集成测试 │ └── conftest.py # 共享 fixtures ├── filesystems/ │ ├── __init__.py │ ├── test_google_drive.py │ └── conftest.py └── queues/ ├── __init__.py └── test_redis_queue.py对照当前仓库的实际布局见 unstract/connectors/tests可以看到这套约定是真实生效的顶层有 test_connectorkit.py用于验证连接器注册表Connectorkittests/databases/ 下目前包含test_bigquery_db.py、test_mariadb.py、test_sql_safety.pytests/filesystems/ 下包含test_miniofs.py、test_box_fs.py、test_google_drive_fs.py、test_sharepoint_fs.py、test_zs_dropbox_fs.py等文档树中的test_postgresql.py属于该约定下的示例命名当前仓库尚未提供对应文件但命名与目录约定是一致的。从 test_connectorkit.py 的源码可以看到注册表测试的具体形态它通过Connectorkit()单例按名称取连接器类并断言其get_connector_mode()与ConnectorMode枚举DATABASE/FILE_SYSTEM匹配。这与参考文档中“静态元数据方法不需要实例化即可测试”的原则完全一致。Mock 测试模板数据库连接器参考文档的核心是一套始终可以运行、不依赖任何外部服务的 Mock 测试模板。下面按文档原样给出模板骨架占位符含义见 test_mock.py.template{ClassName}为 PascalCase 类名、{connector_name}为小写连接器名、{connection_lib}为连接库名如psycopg2import unittest from unittest.mock import patch, Mock, MagicMock import os from unstract.connectors.databases.{connector_name}.{connector_name} import {ClassName} class Test{ClassName}(unittest.TestCase): Mock-based tests for {ClassName} connector. def setUp(self): Set up test fixtures. self.basic_config { host: localhost, port: 5432, database: testdb, user: testuser, password: testpass, } self.url_config { connection_url: postgresql://user:passlocalhost:5432/db } def test_static_methods(self): Test static connector metadata methods. # These dont require instantiation self.assertIsNotNone({ClassName}.get_id()) self.assertIsNotNone({ClassName}.get_name()) self.assertIsNotNone({ClassName}.get_description()) self.assertIsNotNone({ClassName}.get_icon()) self.assertIsNotNone({ClassName}.get_json_schema()) self.assertIsInstance({ClassName}.can_write(), bool) self.assertIsInstance({ClassName}.can_read(), bool) self.assertIsInstance({ClassName}.requires_oauth(), bool) def test_connector_id_format(self): Test connector ID follows expected format. connector_id {ClassName}.get_id() self.assertIn(|, connector_id) name, uuid connector_id.split(|, 1) self.assertTrue(len(uuid) 0) def test_json_schema_valid(self): Test JSON schema is valid JSON. import json schema_str {ClassName}.get_json_schema() schema json.loads(schema_str) self.assertIn(title, schema) self.assertEqual(schema[type], object) patch(unstract.connectors.databases.{connector_name}.{connector_name}.{connection_lib}.connect) def test_initialization_with_params(self, mock_connect): Test connector initializes with individual parameters. mock_connect.return_value Mock() connector {ClassName}(self.basic_config) self.assertEqual(connector.host, localhost) self.assertEqual(connector.port, 5432) self.assertEqual(connector.database, testdb) patch(unstract.connectors.databases.{connector_name}.{connector_name}.{connection_lib}.connect) def test_initialization_with_url(self, mock_connect): Test connector initializes with connection URL. mock_connect.return_value Mock() connector {ClassName}(self.url_config) self.assertEqual(connector.connection_url, self.url_config[connection_url]) patch(unstract.connectors.databases.{connector_name}.{connector_name}.{connection_lib}.connect) def test_get_engine(self, mock_connect): Test get_engine returns connection. mock_connection Mock() mock_connect.return_value mock_connection connector {ClassName}(self.basic_config) engine connector.get_engine() mock_connect.assert_called_once() self.assertEqual(engine, mock_connection) patch(unstract.connectors.databases.{connector_name}.{connector_name}.{connection_lib}.connect) def test_test_credentials_success(self, mock_connect): Test test_credentials returns True on success. mock_connection Mock() mock_connect.return_value mock_connection connector {ClassName}(self.basic_config) result connector.test_credentials() self.assertTrue(result) patch(unstract.connectors.databases.{connector_name}.{connector_name}.{connection_lib}.connect) def test_test_credentials_failure(self, mock_connect): Test test_credentials raises on failure. mock_connect.side_effect Exception(Connection failed) connector {ClassName}(self.basic_config) from unstract.connectors.exceptions import ConnectorError with self.assertRaises(ConnectorError): connector.test_credentials() patch(unstract.connectors.databases.{connector_name}.{connector_name}.{connection_lib}.connect) def test_execute_query(self, mock_connect): Test query execution. mock_cursor Mock() mock_cursor.fetchall.return_value [(row1,), (row2,)] mock_connection Mock() mock_connection.cursor.return_value.__enter__ Mock(return_valuemock_cursor) mock_connection.cursor.return_value.__exit__ Mock(return_valueFalse) mock_connect.return_value mock_connection connector {ClassName}(self.basic_config) result connector.execute(SELECT * FROM test) mock_cursor.execute.assert_called_with(SELECT * FROM test) def test_sql_to_db_mapping_string(self): Test type mapping for strings. connector {ClassName}.__new__({ClassName}) result connector.sql_to_db_mapping(test string) self.assertIn(result.upper(), [TEXT, VARCHAR, STRING]) def test_sql_to_db_mapping_integer(self): Test type mapping for integers. connector {ClassName}.__new__({ClassName}) result connector.sql_to_db_mapping(42) self.assertIn(result.upper(), [INTEGER, INT, BIGINT, INT64]) def test_sql_to_db_mapping_dict(self): Test type mapping for dictionaries (JSON). connector {ClassName}.__new__({ClassName}) result connector.sql_to_db_mapping({key: value}) self.assertIn(result.upper(), [JSON, JSONB, VARIANT, STRING]) if __name__ __main__: unittest.main()模板各部分与源码契约的对应关系模板中的每个断言都不是凭空而来而是对应基类 UnstractDB 的公开契约。结合源码可以逐条印证1. 静态元数据方法无需实例化。get_id()、get_name()、get_description()、get_icon()、get_json_schema()、can_write()、can_read()、requires_oauth()在基类中全部是staticmethod见 unstract_db.py因此可以在不创建连接器实例、不建立任何连接的前提下完成断言。以 postgresql.py 为例get_id()返回postgresql|6db35f45-be11-4fd5-80c5-85c48183afbb这样的短名|UUID字符串get_json_schema()则读取static/json_schema.json文件内容——这正是模板中“ID 含|分隔符”“schema 是合法 JSON 且含title、type: object”两个断言的依据。2. 初始化测试的两种配置形态。模板同时覆盖basic_config离散参数与url_config连接串。这与真实实现一致PostgreSQL.init会在两者都缺失时抛出ValueError(Either ConnectionURL or connection parameters must be provided.)测试中分别断言connector.host/connector.connection_url即可锁定这一行为。3.test_credentials的成功/失败路径。基类实现unstract_db.py是“能连上就返回True连不上就把底层异常包装成ConnectorError抛出”。ConnectorError定义在 exceptions.py因此模板中mock_connect.side_effect Exception(...)assertRaises(ConnectorError)的组合精确复现了这条异常链路。4.execute()的 cursor 上下文管理器约定。基类 execute() 使用with self.get_engine().cursor() as cursor并在无参时直接cursor.execute(query)后fetchall()。这就是为什么模板里要用mock_connection.cursor.return_value.__enter__ Mock(return_valuemock_cursor)这种写法来模拟上下文管理器——它对应的是 Python DB-API 的with语义而不是普通方法调用。5.sql_to_db_mapping的类型映射断言。模板对 string / integer / dict 三类输入给出了一组“可接受的类型名”白名单。这里需要注意不同连接器的实际映射不同应以被测连接器自己的实现为准。例如 PostgreSQL 的 sql_to_db_mappingstr映射TEXT、int映射INTEGER而dict/list默认映射TEXT仅当列名以_v2结尾时才映射JSONB。因此在为 PostgreSQL 写该断言时dict 用例应把TEXT加入允许列表——模板给出的白名单是“通用起点”落地时务必与源码核对。6. patch 路径的确定。模板统一 patchunstract.connectors.databases.{connector_name}.{connector_name}.{connection_lib}.connect。注意路径中出现了两次连接器名第一次是模块包路径第二次是被 patch 模块所在文件的模块名。以 PostgreSQL 为例源码 postgresql.py 直接调用psycopg2.connect(**conn_params)因此正确的 patch 目标是unstract.connectors.databases.postgresql.postgresql.psycopg2.connect。patch 在“被使用处”的命名空间下这是unittest.mock的标准要求也是模板固定该格式的原因。集成测试模板依赖真实服务集成测试与 Mock 测试的关键区别在于没有配置好真实服务时它必须自动跳过而不是失败。参考文档给出的模板与 test_integration.py.template 一致核心结构如下import unittest import os from unstract.connectors.databases.{connector_name}.{connector_name} import {ClassName} unittest.skipUnless( os.getenv({CONNECTOR_PREFIX}_HOST), Integration tests require {CONNECTOR_PREFIX}_* environment variables ) class Test{ClassName}Integration(unittest.TestCase): Integration tests for {ClassName} - requires real database. classmethod def setUpClass(cls): Set up integration test configuration from environment. cls.config { host: os.getenv({CONNECTOR_PREFIX}_HOST), port: os.getenv({CONNECTOR_PREFIX}_PORT, 5432), database: os.getenv({CONNECTOR_PREFIX}_DATABASE), user: os.getenv({CONNECTOR_PREFIX}_USER), password: os.getenv({CONNECTOR_PREFIX}_PASSWORD), } # Validate all required vars present missing [k for k, v in cls.config.items() if not v] if missing: raise unittest.SkipTest(fMissing env vars: {missing}) def test_real_connection(self): Test connecting to real database. connector {ClassName}(self.config) self.assertTrue(connector.test_credentials()) def test_real_query(self): Test executing query on real database. connector {ClassName}(self.config) engine connector.get_engine() with engine.cursor() as cursor: cursor.execute(SELECT 1 AS test) result cursor.fetchone() self.assertEqual(result[0], 1) engine.close() def test_real_table_operations(self): Test table operations on real database. connector {ClassName}(self.config) engine connector.get_engine() try: with engine.cursor() as cursor: # Create test table cursor.execute( CREATE TABLE IF NOT EXISTS _unstract_test_table ( id SERIAL PRIMARY KEY, name TEXT, created_at TIMESTAMP DEFAULT NOW() ) ) # Insert data cursor.execute( INSERT INTO _unstract_test_table (name) VALUES (%s), (test_value,) ) # Query data cursor.execute(SELECT name FROM _unstract_test_table WHERE name %s, (test_value,)) result cursor.fetchone() self.assertEqual(result[0], test_value) engine.commit() finally: # Cleanup with engine.cursor() as cursor: cursor.execute(DROP TABLE IF EXISTS _unstract_test_table) engine.commit() engine.close() if __name__ __main__: unittest.main()这个模板有三个值得注意的工程细节双层跳过机制类级unittest.skipUnless在环境变量缺失时整体跳过setUpClass中再对“有 host 但缺 user/password”这类部分配置的情况抛出unittest.SkipTest避免把“配置不全”误报成测试失败。固定命名的临时表_unstract_test_table配合try/finally确保即使断言失败也会执行DROP TABLE清理不会污染真实库。参数化 SQL%s占位符 元组参数而不是字符串拼接——这也与仓库中 sql_safety.py 所体现的“标识符安全引用 参数化查询”安全基调一致。当前仓库中的真实集成测试印证了同一模式。例如 test_miniofs.py 中的test_minio同时使用了pytest.mark.integration标记和unittest.skipUnless(os.environ.get(MINIO_ACCESS_KEY_ID) and os.environ.get(MINIO_SECRET_ACCESS_KEY), ...)未配置 MinIO 凭据时自动跳过配置后则对真实端点执行get_fsspec_fs()与 bucket 列表操作。文件系统连接器 Mock 测试模板文件系统类连接器基于 fsspec 抽象因此模板的 mock 对象从“数据库连接”换成了“fsspec 文件系统类”{fsspec_class}如S3FileSystem。参考文档给出的完整模板import unittest from unittest.mock import patch, Mock, MagicMock from io import BytesIO from unstract.connectors.filesystems.{connector_name}.{connector_name} import {ClassName} class Test{ClassName}(unittest.TestCase): Mock-based tests for {ClassName} filesystem connector. def setUp(self): self.config { access_key: test_key, secret_key: test_secret, bucket: test-bucket, endpoint_url: https://s3.example.com, } def test_static_methods(self): Test static connector metadata. self.assertIsNotNone({ClassName}.get_id()) self.assertIsNotNone({ClassName}.get_name()) self.assertEqual({ClassName}.get_connector_mode().value, FILESYSTEM) patch(unstract.connectors.filesystems.{connector_name}.{connector_name}.{fsspec_class}) def test_get_fsspec_fs(self, mock_fs_class): Test fsspec filesystem creation. mock_fs Mock() mock_fs_class.return_value mock_fs connector {ClassName}(self.config) fs connector.get_fsspec_fs() self.assertEqual(fs, mock_fs) patch(unstract.connectors.filesystems.{connector_name}.{connector_name}.{fsspec_class}) def test_test_credentials_success(self, mock_fs_class): Test credentials validation success. mock_fs Mock() mock_fs.ls.return_value [file1.txt, file2.txt] mock_fs_class.return_value mock_fs connector {ClassName}(self.config) result connector.test_credentials() self.assertTrue(result) patch(unstract.connectors.filesystems.{connector_name}.{connector_name}.{fsspec_class}) def test_test_credentials_failure(self, mock_fs_class): Test credentials validation failure. mock_fs_class.side_effect Exception(Access denied) connector {ClassName}(self.config) from unstract.connectors.exceptions import ConnectorError with self.assertRaises(ConnectorError): connector.test_credentials() def test_extract_metadata_file_hash(self): Test file hash extraction from metadata. connector {ClassName}.__new__({ClassName}) metadata {ETag: abc123, size: 1024} result connector.extract_metadata_file_hash(metadata) self.assertEqual(result, abc123) def test_is_dir_by_metadata(self): Test directory detection from metadata. connector {ClassName}.__new__({ClassName}) dir_metadata {type: directory} file_metadata {type: file} self.assertTrue(connector.is_dir_by_metadata(dir_metadata)) self.assertFalse(connector.is_dir_by_metadata(file_metadata)) if __name__ __main__: unittest.main()该模板覆盖的文件系统连接器四大契约方法与 SKILL 文档中对 fsspec 封装型连接器的要求见 connector-ops SKILL 的 “Approach 1: Wrapping an existing fsspec provider”一一对应get_fsspec_fs()—— 返回 provider 的文件系统实例如S3FileSystemmock 时直接 patch 该类构造器即可test_credentials()—— 成功路径通常靠一次ls()调用验证访问权限失败路径由异常转换为ConnectorError仓库中对应的 FS 异常族有 FSAccessDeniedError / PermissionDeniedError 等extract_metadata_file_hash()—— 从 listing 元数据中提取文件哈希如 S3 的ETag是增量同步/去重的关键is_dir_by_metadata()—— 判断元数据条目是目录还是文件。值得强调的是后两个方法在模板中通过{ClassName}.__new__({ClassName})跳过__init__直接实例化对象来测试——因为这些方法是纯函数式的元数据解析不需要真实的凭据或网络。同样的技巧在 test_miniofs.py 的TestAccessFilteredS3FileSystem中被用于绕过需要网络/事件循环的__init__从而对 bucket 访问过滤逻辑做纯单元级验证。队列连接器 Mock 测试模板队列类连接器的契约是enqueue/dequeue/peek三个方法模板围绕消息入队、出队与窥视三条路径展开import unittest from unittest.mock import patch, Mock from unstract.connectors.queues.{connector_name}.{connector_name} import {ClassName} class Test{ClassName}(unittest.TestCase): Mock-based tests for {ClassName} queue connector. def setUp(self): self.config { host: localhost, port: 6379, password: testpass, } def test_static_methods(self): Test static connector metadata. self.assertIsNotNone({ClassName}.get_id()) self.assertEqual({ClassName}.get_connector_mode().value, MANUAL_REVIEW) patch(unstract.connectors.queues.{connector_name}.{connector_name}.{queue_lib}) def test_enqueue(self, mock_client): Test message enqueueing. mock_connection Mock() mock_client.return_value mock_connection connector {ClassName}(self.config) connector.enqueue(test_queue, test_message) # Assert enqueue was called mock_connection.lpush.assert_called() # or appropriate method patch(unstract.connectors.queues.{connector_name}.{connector_name}.{queue_lib}) def test_dequeue(self, mock_client): Test message dequeueing. mock_connection Mock() mock_connection.brpop.return_value (test_queue, btest_message) mock_client.return_value mock_connection connector {ClassName}(self.config) result connector.dequeue(test_queue) self.assertEqual(result, test_message) patch(unstract.connectors.queues.{connector_name}.{connector_name}.{queue_lib}) def test_peek(self, mock_client): Test message peeking. mock_connection Mock() mock_connection.lindex.return_value btest_message mock_client.return_value mock_connection connector {ClassName}(self.config) result connector.peek(test_queue) self.assertEqual(result, test_message) if __name__ __main__: unittest.main()与数据库模板相比队列模板有两处差异值得留意patch 的是客户端类本身{queue_lib}如redis.Redis而不是某个connect函数——因为 redis 等客户端库的典型用法是“构造客户端对象后再调用命令方法”返回模式匹配底层命令模板中dequeue断言brpop返回(queue, bytes)二元组、peek断言lindex返回 bytes这体现了 Redis list 作为队列时阻塞弹出与按索引窥视的真实命令语义写具体连接器测试时应按实际客户端命令调整模板注释中也提醒 “or appropriate method”。get_connector_mode().value MANUAL_REVIEW这一断言对应仓库的连接器模式枚举设计队列连接器在 Unstract 中承担的是“人工审核”通道角色见 SKILL.md 中的架构总表。共享 Fixturesconftest.py当同一目录下多个测试文件需要相同配置时参考文档建议在conftest.py中定义 pytest fixtureimport pytest import os pytest.fixture def mock_db_config(): Standard database test configuration. return { host: localhost, port: 5432, database: testdb, user: testuser, password: testpass, } pytest.fixture def mock_fs_config(): Standard filesystem test configuration. return { access_key: test_key, secret_key: test_secret, bucket: test-bucket, } pytest.fixture def integration_db_config(): Integration test database configuration from environment. config { host: os.getenv(TEST_DB_HOST), port: os.getenv(TEST_DB_PORT, 5432), database: os.getenv(TEST_DB_DATABASE), user: os.getenv(TEST_DB_USER), password: os.getenv(TEST_DB_PASSWORD), } if not all(config.values()): pytest.skip(Integration test requires TEST_DB_* environment variables) return config这里的设计要点mock_db_config/mock_fs_config提供固定假值保证 Mock 测试在任何机器上都能稳定复现integration_db_config从TEST_DB_*环境变量读取配置并在缺项时调用pytest.skip(...)优雅跳过——注意它与上一节unittest.skipUnless是等价的两套机制fixture 方式更适合 pytest 风格写法每个类型子目录tests/databases/、tests/filesystems/各放一份conftest.py与文档给出的目录结构保持一致。仓库顶层的 tests/conftest.py 则展示了 conftest 的另一半职责在pytest_configure钩子中从unstract/connectors/.env加载环境变量load_dotenv。两者配合后集成测试的配置链路就是.env文件或真实环境变量→ conftest 加载 → fixture /os.getenv读取 → skip 判定。这与参考文档“Running Tests”一节中先export再运行的做法互为补充本地手动跑集成测试用exportCI 或.env场景由 conftest 自动完成。运行测试参考文档给出的完整运行方式在unstract/connectors包目录下执行该目录带有独立的 pyproject.toml 与uv.lock因此使用uv run在隔离环境中执行# Run all mock tests (always works) cd unstract/connectors uv run pytest tests/ -v --ignoretests/**/test_*_integration.py # Run specific connector tests uv run pytest tests/databases/test_postgresql.py -v # Run integration tests (requires env vars) export POSTGRESQL_HOSTlocalhost export POSTGRESQL_PORT5432 export POSTGRESQL_DATABASEtestdb export POSTGRESQL_USERtestuser export POSTGRESQL_PASSWORDtestpass uv run pytest tests/databases/test_postgresql_integration.py -v # Run with coverage uv run pytest tests/ -v --covsrc/unstract/connectors --cov-reporthtml要点说明日常 CI/开发只跑 Mock 测试通过--ignoretests/**/test_*_integration.py排除集成文件。即使不排除集成测试自身的skipUnless兜底也会跳过但显式 ignore 能省去收集阶段的时间集成测试的环境变量前缀{CONNECTOR_PREFIX}_*与模板中的类保持一致如上例的POSTGRESQL_*PORT类变量允许有默认值覆盖率命令中--covsrc/unstract/connectors的包路径对应 unstract/connectors/src/unstract/connectors 的实际源码布局src布局。常用断言速查参考文档最后汇总了一组高频断言模式可作为检查清单# Connector ID format self.assertRegex(connector.get_id(), r^[a-z_]\|[a-f0-9-]$) # JSON schema validity import json schema json.loads(connector.get_json_schema()) self.assertIn(title, schema) self.assertIn(type, schema) # Connection returns correct type from psycopg2.extensions import connection self.assertIsInstance(connector.get_engine(), connection) # Exception handling from unstract.connectors.exceptions import ConnectorError with self.assertRaises(ConnectorError) as ctx: connector.test_credentials() self.assertIn(expected message, str(ctx.exception))逐条对应到仓库事实ID 正则^[a-z_]\|[a-f0-9-]$精确刻画了“小写短名 | UUID”的格式约束可对照 postgresql.py 的 get_id() 验证连接类型断言以psycopg2.extensions.connection为例与 PostgreSQL.get_engine() 的返回类型标注一致其他数据库应替换为各自驱动的连接类型如 MySQL 用mysql.connector的连接对象异常消息断言assertRaises后检查str(ctx.exception)适合验证ConnectorError携带的原始错误信息是否完整透传。仓库中 test_bigquery_db.py 展示了更精细的用法它 mock 出 Google API 的Forbidden/NotFound异常断言连接器抛出的BigQueryForbiddenException.detail既包含默认文案Access forbidden in bigquery、Please check your permissions又包含真实错误详情403 Billing has not been enabled...甚至验证了 detail 为空时不输出多余的 Details: 段落——这是“异常路径也要断言消息内容”的完整示范。实践总结结合参考文档与 connector-ops SKILL 的 Important Notes为 Unstract 连接器写测试时应遵守的纪律可以归纳为Mock 测试永远零外部依赖patch 连接库的构造/连接入口覆盖静态元数据、初始化、连接、凭据校验、核心操作与类型映射六类契约集成测试必须可跳过统一使用{CONNECTOR}_*环境变量前缀 skipUnless或pytest.skip双重保护配合.env加载见 tests/conftest.py实现“本地有服务就跑没有就跳”异常路径断言到消息级别ConnectorError及 FS 子类异常是连接器对外的错误契约用assertRaises上下文管理器检查detail/str(exception)内容断言白名单与实现核对模板给出的类型映射白名单是起点落地前必须与目标连接器的sql_to_db_mapping/extract_metadata_file_hash等实现逐一比对避免“模板通过、实现不符”的假阳性。按上述模式任何一个新连接器在合入unstract/connectors时都应同时携带tests/{type}/test_{name}.pyMock与tests/{type}/test_{name}_integration.py集成两个文件使uv run pytest tests/ -v --ignoretests/**/test_*_integration.py在任意环境中稳定通过。【免费下载链接】unstractLLM-Driven Extraction of Unstructured Data — Built for API Deployments ETL Pipeline Workflows项目地址: https://gitcode.com/GitHub_Trending/un/unstract创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考