feat: +unit test

fixbug: PYTHONPATH

fixbug: unit test

tests/metagpt/roles/test_assistant.py

@@ -41,7 +41,7 @@ async def test_run():
                     {"content": "The one who eaten a poison apple.", "role": "assistant"},
                 ],
                 "knowledge": [{"content": "tulin is a scientist."}],
-                "last_talk": "what's apple?",
+                "last_talk": "Do you have a poison apple?",
             },
             "language": "English",
             "agent_description": "chatterbox",

tests/metagpt/tools/test_hello.py

@@ -12,11 +12,16 @@ from pathlib import Path
 import pytest
 import requests
 
+from metagpt.config import CONFIG
+
 
 @pytest.mark.asyncio
 async def test_hello():
-    script_pathname = Path(__file__).parent / "../../../metagpt/tools/hello.py"
-    process = subprocess.Popen(["python", str(script_pathname)])
+    workdir = Path(__file__).parent.parent.parent.parent
+    script_pathname = workdir / "metagpt/tools/hello.py"
+    env = CONFIG.new_environ()
+    env["PYTHONPATH"] = str(workdir) + ":" + env.get("PYTHONPATH", "")
+    process = subprocess.Popen(["python", str(script_pathname)], cwd=workdir, env=env)
     await asyncio.sleep(5)
 
     url = "http://localhost:8082/openapi/greeting/dave"

tests/metagpt/tools/test_metagpt_oas3_api_svc.py

@@ -12,11 +12,16 @@ from pathlib import Path
 import pytest
 import requests
 
+from metagpt.config import CONFIG
+
 
 @pytest.mark.asyncio
 async def test_oas2_svc():
-    script_pathname = Path(__file__).parent / "../../../metagpt/tools/metagpt_oas3_api_svc.py"
-    process = subprocess.Popen(["python", str(script_pathname)])
+    workdir = Path(__file__).parent.parent.parent.parent
+    script_pathname = workdir / "metagpt/tools/metagpt_oas3_api_svc.py"
+    env = CONFIG.new_environ()
+    env["PYTHONPATH"] = str(workdir) + ":" + env.get("PYTHONPATH", "")
+    process = subprocess.Popen(["python", str(script_pathname)], cwd=str(workdir), env=env)
     await asyncio.sleep(5)
 
     url = "http://localhost:8080/openapi/greeting/dave"

tests/metagpt/tools/test_ut_writer.py

@@ -9,34 +9,34 @@ from pathlib import Path
 
 import pytest
 
-from metagpt.const import API_QUESTIONS_PATH, SWAGGER_PATH, UT_PY_PATH
+from metagpt.config import CONFIG
+from metagpt.const import API_QUESTIONS_PATH, UT_PY_PATH
 from metagpt.tools.ut_writer import YFT_PROMPT_PREFIX, UTGenerator
 
 
 class TestUTWriter:
-    def test_api_to_ut_sample(self):
+    @pytest.mark.asyncio
+    async def test_api_to_ut_sample(self):
         # Prerequisites
-        swagger_file = SWAGGER_PATH / "yft_swaggerApi.json"
+        swagger_file = Path(__file__).parent / "../../data/ut_writer/yft_swaggerApi.json"
         assert swagger_file.exists()
+        assert CONFIG.OPENAI_API_KEY and CONFIG.OPENAI_API_KEY != "YOUR_API_KEY"
+        assert not CONFIG.OPENAI_API_TYPE
+        assert CONFIG.OPENAI_API_MODEL
 
-        tags = ["测试"]  # "智能合同导入", "律师审查", "ai合同审查", "草拟合同&律师在线审查", "合同审批", "履约管理", "签约公司"]
+        tags = ["测试", "作业"]
         # 这里在文件中手动加入了两个测试标签的API
 
         utg = UTGenerator(
-            swagger_file=swagger_file,
+            swagger_file=str(swagger_file),
             ut_py_path=UT_PY_PATH,
             questions_path=API_QUESTIONS_PATH,
             template_prefix=YFT_PROMPT_PREFIX,
         )
-        ret = utg.generate_ut(include_tags=tags)
+        ret = await utg.generate_ut(include_tags=tags)
         # 后续加入对文件生成内容与数量的检验
         assert ret
 
-        pathname = Path(__file__).with_suffix(".tmp")
-        utg.ask_gpt_and_save(question="question", tag="tag", fname=str(pathname))
-        assert pathname.exists()
-        pathname.unlink(missing_ok=True)
-
 
 if __name__ == "__main__":
     pytest.main([__file__, "-s"])

tests/metagpt/utils/test_common.py

@@ -6,11 +6,13 @@
 @File    : test_common.py
 @Modified by: mashenquan, 2023/11/21. Add unit tests.
 """
-
+import importlib
 import os
 import platform
+from pathlib import Path
 from typing import Any, Set
 
+import aiofiles
 import pytest
 from pydantic import BaseModel
 
@@ -18,7 +20,20 @@ from metagpt.actions import RunCode
 from metagpt.const import get_metagpt_root
 from metagpt.roles.tutorial_assistant import TutorialAssistant
 from metagpt.schema import Message
-from metagpt.utils.common import any_to_str, any_to_str_set, check_cmd_exists
+from metagpt.utils.common import (
+    NoMoneyException,
+    OutputParser,
+    any_to_str,
+    any_to_str_set,
+    check_cmd_exists,
+    concat_namespace,
+    import_class_inst,
+    parse_recipient,
+    print_members,
+    read_file_block,
+    read_json_file,
+    require_python_version,
+)
 
 
 class TestGetProjectRoot:
@@ -96,6 +111,65 @@ class TestGetProjectRoot:
             else:
                 assert result != 0
 
+    @pytest.mark.parametrize(("filename", "want"), [("1.md", "File list"), ("2.md", "Language"), ("3.md", "# TODOs")])
+    @pytest.mark.asyncio
+    async def test_parse_data_exception(self, filename, want):
+        pathname = Path(__file__).parent.parent.parent / "data/output_parser" / filename
+        assert pathname.exists()
+        async with aiofiles.open(str(pathname), mode="r") as reader:
+            data = await reader.read()
+
+        result = OutputParser.parse_data(data=data)
+        assert want in result
+
+    @pytest.mark.parametrize(
+        ("ver", "want", "err"), [((1, 2, 3, 4), False, True), ((2, 3, 9), True, False), ((3, 10, 18), False, False)]
+    )
+    def test_require_python_version(self, ver, want, err):
+        try:
+            res = require_python_version(ver)
+            assert res == want
+        except ValueError:
+            assert err
+
+    def test_no_money_exception(self):
+        val = NoMoneyException(3.10)
+        assert "Amount required:" in str(val)
+
+    @pytest.mark.parametrize("module_path", ["tests.metagpt.utils.test_common"])
+    def test_print_members(self, module_path):
+        module = importlib.import_module(module_path)
+        with pytest.raises(Exception) as info:
+            print_members(module)
+            assert info is None
+
+    @pytest.mark.parametrize(
+        ("words", "want"), [("", ""), ("## Send To: Engineer", "Engineer"), ("Send To: \nNone", "None")]
+    )
+    def test_parse_recipient(self, words, want):
+        res = parse_recipient(words)
+        assert want == res
+
+    def test_concat_namespace(self):
+        assert concat_namespace("a", "b", "c") == "a:b:c"
+        assert concat_namespace("a", "b", "c", "e") == "a:b:c:e"
+        assert concat_namespace("a", "b", "c", "e", "f") == "a:b:c:e:f"
+
+    def test_read_json_file(self):
+        assert read_json_file(str(Path(__file__).parent / "../../data/ut_writer/yft_swaggerApi.json"), encoding="utf-8")
+        with pytest.raises(FileNotFoundError):
+            read_json_file("not_exists_file", encoding="utf-8")
+        with pytest.raises(ValueError):
+            read_json_file(__file__, encoding="utf-8")
+
+    def test_import_class_inst(self):
+        rc = import_class_inst("RunCode", "metagpt.actions.run_code", name="X")
+        assert rc.name == "X"
+
+    @pytest.mark.asyncio
+    async def test_read_file_block(self):
+        assert await read_file_block(filename=__file__, lineno=6, end_lineno=6) == "@File    : test_common.py\n"
+
 
 if __name__ == "__main__":
     pytest.main([__file__, "-s"])

tests/metagpt/utils/test_cost_manager.py

@@ -0,0 +1,32 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+"""
+@Time    : 2023/12/27
+@Author  : mashenquan
+@File    : test_cost_manager.py
+"""
+import pytest
+
+from metagpt.utils.cost_manager import CostManager
+
+
+def test_cost_manager():
+    cm = CostManager(total_budget=20)
+    cm.update_cost(prompt_tokens=1000, completion_tokens=100, model="gpt-4-1106-preview")
+    assert cm.get_total_prompt_tokens() == 1000
+    assert cm.get_total_completion_tokens() == 100
+    assert cm.get_total_cost() == 0.013
+    cm.update_cost(prompt_tokens=100, completion_tokens=10, model="gpt-4-1106-preview")
+    assert cm.get_total_prompt_tokens() == 1100
+    assert cm.get_total_completion_tokens() == 110
+    assert cm.get_total_cost() == 0.0143
+    cost = cm.get_costs()
+    assert cost
+    assert cost.total_cost == cm.get_total_cost()
+    assert cost.total_prompt_tokens == cm.get_total_prompt_tokens()
+    assert cost.total_completion_tokens == cm.get_total_completion_tokens()
+    assert cost.total_budget == 20
+
+
+if __name__ == "__main__":
+    pytest.main([__file__, "-s"])

tests/metagpt/utils/test_file.py

@@ -23,3 +23,13 @@ async def test_write_and_read_file(root_path: Path, filename: str, content: byte
     assert root_path / filename == full_file_name
     file_data = await File.read(full_file_name)
     assert file_data.decode("utf-8") == content
+
+
+@pytest.mark.asyncio
+async def test_read_chunk():
+    val = await File.read(file_path=__file__, chunk_size=10)
+    assert val
+
+
+if __name__ == "__main__":
+    pytest.main([__file__, "-s"])

tests/metagpt/utils/test_file_repository.py

@@ -33,20 +33,22 @@ async def test_file_repo():
     assert file_repo.workdir == full_path
     assert file_repo.workdir.exists()
     await file_repo.save("a.txt", "AAA")
-    await file_repo.save("b.txt", "BBB", ["a.txt"])
+    await file_repo.save("b.txt", "BBB", [str(full_path / "a.txt"), f"{file_repo_path}/c.txt"])
     doc = await file_repo.get("a.txt")
     assert "AAA" == doc.content
     doc = await file_repo.get("b.txt")
     assert "BBB" == doc.content
-    assert {"a.txt"} == await file_repo.get_dependency("b.txt")
+    assert {f"{file_repo_path}/a.txt", f"{file_repo_path}/c.txt"} == await file_repo.get_dependency("b.txt")
     assert {"a.txt": ChangeType.UNTRACTED, "b.txt": ChangeType.UNTRACTED} == file_repo.changed_files
-    assert {"a.txt"} == await file_repo.get_changed_dependency("b.txt")
+    assert {f"{file_repo_path}/a.txt"} == await file_repo.get_changed_dependency("b.txt")
     await file_repo.save("d/e.txt", "EEE")
     assert ["d/e.txt"] == file_repo.get_change_dir_files("d")
     assert set(file_repo.all_files) == {"a.txt", "b.txt", "d/e.txt"}
     await file_repo.delete("d/e.txt")
     await file_repo.delete("d/e.txt")  # delete twice
     assert set(file_repo.all_files) == {"a.txt", "b.txt"}
+    await file_repo.delete("b.txt")
+    assert set(file_repo.all_files) == {"a.txt"}
 
     git_repo.delete_repository()
 

tests/metagpt/utils/test_git_repository.py

@@ -61,6 +61,11 @@ async def test_git():
 
     assert repo.status
 
+    exist_dir = repo.workdir / "git4"
+    exist_dir.mkdir(parents=True, exist_ok=True)
+    repo.rename_root("git4")
+    assert repo.workdir.name == "git4"
+
     repo.delete_repository()
     assert not local_path.exists()
 
@@ -80,6 +85,9 @@ async def test_git1():
     all_files = repo1.get_files(relative_path=".", filter_ignored=True)
     assert "__pycache__/a.pyc" not in all_files
 
+    res = repo1.filter_gitignore(filenames=["snake_game/snake_game/__pycache__", "snake_game/snake_game/game.py"])
+    assert res == ["snake_game/snake_game/game.py"]
+
     repo1.delete_repository()
     assert not local_path.exists()
 
@@ -99,5 +107,20 @@ async def test_dependency_file():
     assert not dependancy_file.exists
 
 
+@pytest.mark.asyncio
+async def test_git_open():
+    local_path = Path(__file__).parent / "git3"
+    local_path.mkdir(exist_ok=True, parents=True)
+
+    assert not GitRepository.is_git_dir(local_path)
+    repo = GitRepository()
+    repo.open(local_path, auto_init=False)
+    assert not repo.is_valid
+    assert not repo.status
+    assert not repo.workdir
+
+    shutil.rmtree(path=str(local_path), ignore_errors=True)
+
+
 if __name__ == "__main__":
     pytest.main([__file__, "-s"])

tests/metagpt/utils/test_mermaid.py

@@ -0,0 +1,39 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+"""
+@Time    : 2023/12/27
+@Author  : mashenquan
+@File    : test_mermaid.py
+"""
+
+import pytest
+
+from metagpt.config import CONFIG
+from metagpt.utils.common import check_cmd_exists
+from metagpt.utils.mermaid import MMC1, MMC2, mermaid_to_file
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("engine", ["nodejs", "playwright", "pyppeteer", "ink"])
+async def test_mermaid(engine):
+    # Prerequisites
+    # npm install -g @mermaid-js/mermaid-cli
+    assert check_cmd_exists("npm") == 0
+    assert CONFIG.PYPPETEER_EXECUTABLE_PATH
+
+    CONFIG.mermaid_engine = engine
+    save_to = CONFIG.git_repo.workdir / f"{CONFIG.mermaid_engine}/1"
+    await mermaid_to_file(MMC1, save_to)
+    for ext in [".pdf", ".svg", ".png"]:
+        assert save_to.with_suffix(ext).exists()
+        save_to.with_suffix(ext).unlink(missing_ok=True)
+
+    save_to = CONFIG.git_repo.workdir / f"{CONFIG.mermaid_engine}/2"
+    await mermaid_to_file(MMC2, save_to)
+    for ext in [".pdf", ".svg", ".png"]:
+        assert save_to.with_suffix(ext).exists()
+        save_to.with_suffix(ext).unlink(missing_ok=True)
+
+
+if __name__ == "__main__":
+    pytest.main([__file__, "-s"])

tests/metagpt/utils/test_redis.py

@@ -0,0 +1,32 @@
+#!/usr/bin/env python3
+# _*_ coding: utf-8 _*_
+"""
+@Time    : 2023/12/27
+@Author  : mashenquan
+@File    : test_redis.py
+"""
+
+import pytest
+
+from metagpt.config import CONFIG
+from metagpt.utils.redis import Redis
+
+
+@pytest.mark.asyncio
+async def test_redis():
+    # Prerequisites
+    assert CONFIG.REDIS_HOST and CONFIG.REDIS_HOST != "YOUR_REDIS_HOST"
+    assert CONFIG.REDIS_PORT and CONFIG.REDIS_PORT != "YOUR_REDIS_PORT"
+    # assert CONFIG.REDIS_USER
+    assert CONFIG.REDIS_PASSWORD is not None and CONFIG.REDIS_PASSWORD != "YOUR_REDIS_PASSWORD"
+    assert CONFIG.REDIS_DB is not None and CONFIG.REDIS_DB != "YOUR_REDIS_DB_INDEX, str, 0-based"
+
+    conn = Redis()
+    assert not conn.is_valid
+    await conn.set("test", "test", timeout_sec=0)
+    assert await conn.get("test") == b"test"
+    await conn.close()
+
+
+if __name__ == "__main__":
+    pytest.main([__file__, "-s"])

tests/metagpt/utils/test_s3.py

@@ -0,0 +1,54 @@
+#!/usr/bin/env python3
+# _*_ coding: utf-8 _*_
+"""
+@Time    : 2023/12/27
+@Author  : mashenquan
+@File    : test_s3.py
+"""
+import uuid
+from pathlib import Path
+
+import aiofiles
+import pytest
+
+from metagpt.config import CONFIG
+from metagpt.utils.s3 import S3
+
+
+@pytest.mark.asyncio
+async def test_s3():
+    # Prerequisites
+    assert CONFIG.S3_ACCESS_KEY and CONFIG.S3_ACCESS_KEY != "YOUR_S3_ACCESS_KEY"
+    assert CONFIG.S3_SECRET_KEY and CONFIG.S3_SECRET_KEY != "YOUR_S3_SECRET_KEY"
+    assert CONFIG.S3_ENDPOINT_URL and CONFIG.S3_ENDPOINT_URL != "YOUR_S3_ENDPOINT_URL"
+    # assert CONFIG.S3_SECURE: true # true/false
+    assert CONFIG.S3_BUCKET and CONFIG.S3_BUCKET != "YOUR_S3_BUCKET"
+
+    conn = S3()
+    assert conn.is_valid
+    object_name = "unittest.bak"
+    await conn.upload_file(bucket=CONFIG.S3_BUCKET, local_path=__file__, object_name=object_name)
+    pathname = (Path(__file__).parent / uuid.uuid4().hex).with_suffix(".bak")
+    pathname.unlink(missing_ok=True)
+    await conn.download_file(bucket=CONFIG.S3_BUCKET, object_name=object_name, local_path=str(pathname))
+    assert pathname.exists()
+    url = await conn.get_object_url(bucket=CONFIG.S3_BUCKET, object_name=object_name)
+    assert url
+    bin_data = await conn.get_object(bucket=CONFIG.S3_BUCKET, object_name=object_name)
+    assert bin_data
+    async with aiofiles.open(__file__, mode="r", encoding="utf-8") as reader:
+        data = await reader.read()
+    res = await conn.cache(data, ".bak", "script")
+    assert "http" in res
+
+
+@pytest.mark.asyncio
+async def test_s3_no_error():
+    conn = S3()
+    conn.auth_config["aws_secret_access_key"] = ""
+    res = await conn.cache("ABC", ".bak", "script")
+    assert not res
+
+
+if __name__ == "__main__":
+    pytest.main([__file__, "-s"])