Initial import of NLProg
This commit is contained in:
sunguosheng
@@ -0,0 +1,8 @@
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
SRC = ROOT / "src"
|
||||
if str(SRC) not in sys.path:
|
||||
sys.path.insert(0, str(SRC))
|
||||
@@ -0,0 +1,24 @@
|
||||
import unittest
|
||||
|
||||
from nlprog.command_safety import assess_command
|
||||
|
||||
|
||||
class CommandSafetyTests(unittest.TestCase):
|
||||
def test_allows_safe_validation_command(self):
|
||||
assessment = assess_command("python -m compileall src")
|
||||
self.assertEqual(assessment.level, "allow")
|
||||
self.assertFalse(assessment.blocked)
|
||||
|
||||
def test_confirms_package_install(self):
|
||||
assessment = assess_command("pip install requests")
|
||||
self.assertEqual(assessment.level, "confirm")
|
||||
self.assertTrue(assessment.needs_confirmation)
|
||||
|
||||
def test_blocks_format(self):
|
||||
assessment = assess_command("format C:")
|
||||
self.assertEqual(assessment.level, "block")
|
||||
self.assertTrue(assessment.blocked)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,94 @@
|
||||
import tempfile
|
||||
import unittest
|
||||
import os
|
||||
from pathlib import Path
|
||||
from unittest.mock import patch
|
||||
|
||||
import nlprog.config as config
|
||||
|
||||
|
||||
class ConfigTests(unittest.TestCase):
|
||||
def test_parse_config_value(self):
|
||||
self.assertTrue(config.parse_config_value("require_confirmation", "true"))
|
||||
self.assertEqual(config.parse_config_value("json_repair_retries", "3"), 3)
|
||||
self.assertEqual(config.parse_config_value("temperature", "0.5"), 0.5)
|
||||
self.assertIsNone(config.parse_config_value("base_url", "null"))
|
||||
|
||||
def test_set_config_value_uses_temp_config_file(self):
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
config_file = Path(tmp) / "config.json"
|
||||
with patch.object(config, "CONFIG_DIR", Path(tmp)), patch.object(config, "CONFIG_FILE", config_file):
|
||||
config.write_default_config()
|
||||
config.set_config_value("provider", "opencode")
|
||||
data = config.read_config_file()
|
||||
self.assertEqual(data["provider"], "opencode")
|
||||
|
||||
def test_active_model_overrides_loaded_config(self):
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
config_file = Path(tmp) / "config.json"
|
||||
with patch.object(config, "CONFIG_DIR", Path(tmp)):
|
||||
with patch.object(config, "CONFIG_FILE", config_file):
|
||||
with patch.dict(os.environ, {}, clear=True):
|
||||
config.write_default_config()
|
||||
config.add_model(
|
||||
"deepseek",
|
||||
provider="openai-compatible",
|
||||
model="deepseek-chat",
|
||||
base_url="https://api.deepseek.com/v1",
|
||||
api_key_env="DEEPSEEK_API_KEY",
|
||||
activate=True,
|
||||
)
|
||||
|
||||
loaded = config.load_config()
|
||||
|
||||
self.assertEqual(loaded.provider, "openai-compatible")
|
||||
self.assertEqual(loaded.model, "deepseek-chat")
|
||||
self.assertEqual(loaded.base_url, "https://api.deepseek.com/v1")
|
||||
self.assertEqual(loaded.api_key_env, "DEEPSEEK_API_KEY")
|
||||
|
||||
def test_legacy_config_without_active_model_still_loads_top_level_values(self):
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
config_file = Path(tmp) / "config.json"
|
||||
with patch.object(config, "CONFIG_DIR", Path(tmp)):
|
||||
with patch.object(config, "CONFIG_FILE", config_file):
|
||||
with patch.dict(os.environ, {}, clear=True):
|
||||
config.write_config_file(
|
||||
{
|
||||
"provider": "opencode",
|
||||
"model": "",
|
||||
"opencode_command": "opencode",
|
||||
}
|
||||
)
|
||||
|
||||
loaded = config.load_config()
|
||||
active, models = config.list_models()
|
||||
|
||||
self.assertEqual(loaded.provider, "opencode")
|
||||
self.assertEqual(loaded.model, "")
|
||||
self.assertEqual(active, "current")
|
||||
self.assertEqual(models["current"]["provider"], "opencode")
|
||||
|
||||
def test_model_registry_use_and_remove(self):
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
config_file = Path(tmp) / "config.json"
|
||||
with patch.object(config, "CONFIG_DIR", Path(tmp)), patch.object(config, "CONFIG_FILE", config_file):
|
||||
config.write_default_config()
|
||||
config.add_model("codex", provider="opencode", activate=True)
|
||||
config.add_model("mock2", provider="mock", model="mock-model")
|
||||
|
||||
active, models = config.list_models()
|
||||
self.assertEqual(active, "codex")
|
||||
self.assertIn("codex", models)
|
||||
self.assertIn("mock2", models)
|
||||
|
||||
with self.assertRaises(ValueError):
|
||||
config.remove_model("codex")
|
||||
|
||||
config.use_model("mock2")
|
||||
config.remove_model("codex")
|
||||
_, models = config.list_models()
|
||||
self.assertNotIn("codex", models)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,38 @@
|
||||
import unittest
|
||||
|
||||
from nlprog.json_repair import complete_json, parse_json_object
|
||||
from nlprog.llm import Message
|
||||
|
||||
|
||||
class BrokenThenGoodClient:
|
||||
def __init__(self):
|
||||
self.calls = 0
|
||||
|
||||
def complete(self, messages):
|
||||
self.calls += 1
|
||||
if self.calls == 1:
|
||||
return "not json"
|
||||
return '{"final": "repaired"}'
|
||||
|
||||
|
||||
class JsonRepairTests(unittest.TestCase):
|
||||
def test_parse_json_object_strips_fences(self):
|
||||
data, error = parse_json_object('```json\n{"ok": true}\n```')
|
||||
self.assertEqual(error, "")
|
||||
self.assertEqual(data, {"ok": True})
|
||||
|
||||
def test_complete_json_repairs_bad_response(self):
|
||||
result = complete_json(
|
||||
BrokenThenGoodClient(),
|
||||
[Message("system", "test"), Message("user", "return json")],
|
||||
'{"final": "summary"}',
|
||||
lambda data: None if "final" in data else "missing final",
|
||||
max_retries=2,
|
||||
)
|
||||
self.assertEqual(result.data, {"final": "repaired"})
|
||||
self.assertEqual(result.attempts, 2)
|
||||
self.assertTrue(result.repaired)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,46 @@
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
from nlprog.patching import PatchError, apply_patch_to_root, summarize_patch
|
||||
|
||||
|
||||
class PatchingTests(unittest.TestCase):
|
||||
def test_add_update_delete_file(self):
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
root = Path(tmp)
|
||||
add_patch = """*** Begin Patch
|
||||
*** Add File: notes.txt
|
||||
+hello
|
||||
*** End Patch"""
|
||||
self.assertIn("Add notes.txt", summarize_patch(add_patch))
|
||||
self.assertEqual(apply_patch_to_root(root, add_patch, set()), ["Added notes.txt"])
|
||||
self.assertEqual((root / "notes.txt").read_text(encoding="utf-8"), "hello\n")
|
||||
|
||||
update_patch = """*** Begin Patch
|
||||
*** Update File: notes.txt
|
||||
@@
|
||||
-hello
|
||||
+hello world
|
||||
*** End Patch"""
|
||||
self.assertEqual(apply_patch_to_root(root, update_patch, set()), ["Updated notes.txt"])
|
||||
self.assertEqual((root / "notes.txt").read_text(encoding="utf-8"), "hello world\n")
|
||||
|
||||
delete_patch = """*** Begin Patch
|
||||
*** Delete File: notes.txt
|
||||
*** End Patch"""
|
||||
self.assertEqual(apply_patch_to_root(root, delete_patch, set()), ["Deleted notes.txt"])
|
||||
self.assertFalse((root / "notes.txt").exists())
|
||||
|
||||
def test_protected_path_is_rejected(self):
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
patch = """*** Begin Patch
|
||||
*** Add File: __pycache__/blocked.txt
|
||||
+x
|
||||
*** End Patch"""
|
||||
with self.assertRaises(PatchError):
|
||||
apply_patch_to_root(Path(tmp), patch, {"__pycache__"})
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,29 @@
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
from nlprog.project_profile import infer_project, init_project, load_project_profile
|
||||
|
||||
|
||||
class ProjectProfileTests(unittest.TestCase):
|
||||
def test_infers_python_project(self):
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
root = Path(tmp)
|
||||
(root / "pyproject.toml").write_text("[project]\nname='x'\n", encoding="utf-8")
|
||||
(root / "src").mkdir()
|
||||
profile = infer_project(root)
|
||||
self.assertIn("python", profile.project_types)
|
||||
self.assertIn("python -m compileall src", profile.verification_commands)
|
||||
|
||||
def test_init_project_writes_profile_files(self):
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
root = Path(tmp)
|
||||
(root / "pyproject.toml").write_text("[project]\nname='x'\n", encoding="utf-8")
|
||||
created = init_project(root)
|
||||
self.assertEqual(len(created), 3)
|
||||
profile = load_project_profile(root)
|
||||
self.assertIn("pyproject.toml", profile.important_files)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,29 @@
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
from nlprog.run_log import list_run_logs, load_run_log, write_run_log
|
||||
|
||||
|
||||
class Event:
|
||||
def __init__(self, kind, message):
|
||||
self.kind = kind
|
||||
self.message = message
|
||||
|
||||
|
||||
class RunLogTests(unittest.TestCase):
|
||||
def test_write_list_and_load_latest_log(self):
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
root = Path(tmp)
|
||||
path = write_run_log(root, "task", "done", [Event("final", "done")], "2026-01-01T00:00:00")
|
||||
self.assertTrue(path.exists())
|
||||
logs = list_run_logs(root)
|
||||
self.assertEqual(logs, [path])
|
||||
loaded_path, data = load_run_log(root, "latest")
|
||||
self.assertEqual(loaded_path, path)
|
||||
self.assertEqual(data["task"], "task")
|
||||
self.assertEqual(data["final"], "done")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user