import os import subprocess import sys from pathlib import Path import TerraFin.agent.cli as cli_module import TerraFin.interface.server as server_module def _run_repo_python(tmp_path, code: str, *, env: dict[str, str] | None = None) -> subprocess.CompletedProcess[str]: repo_src = Path(__file__).resolve().parents[1] / "src" command = [ sys.executable, "-c", ( "import sys; " f"sys.path.insert(0, {str(repo_src).__repr__()}); " f"{code}" ), ] return subprocess.run( command, cwd=tmp_path, capture_output=True, text=True, check=True, env=env, ) def test_importing_terrafin_does_not_load_dotenv(tmp_path) -> None: env_file = tmp_path / ".env" env_file.write_text("TERRAFIN_TEST_AUTOLOAD=loaded\n", encoding="utf-8") result = _run_repo_python( tmp_path, "import os; import TerraFin; print(os.getenv('TERRAFIN_TEST_AUTOLOAD', ''))", ) assert result.stdout.strip() == "" def test_agent_cli_main_loads_dotenv_from_explicit_entrypoint(monkeypatch, tmp_path) -> None: (tmp_path / ".env").write_text("TERRAFIN_TEST_AGENT_DOTENV=cli-loaded\n", encoding="utf-8") monkeypatch.chdir(tmp_path) monkeypatch.delenv("TERRAFIN_TEST_AGENT_DOTENV", raising=False) class _FakeClient: def resolve(self, query: str) -> dict: return {"query": query} monkeypatch.setattr(cli_module, "_make_client", lambda args: _FakeClient()) assert cli_module.main(["resolve", "AAPL"]) == 0 assert os.getenv("TERRAFIN_TEST_AGENT_DOTENV") == "cli-loaded" def test_server_main_loads_dotenv_from_explicit_entrypoint(monkeypatch, tmp_path) -> None: (tmp_path / ".env").write_text("TERRAFIN_TEST_SERVER_DOTENV=server-loaded\n", encoding="utf-8") monkeypatch.chdir(tmp_path) monkeypatch.delenv("TERRAFIN_TEST_SERVER_DOTENV", raising=False) monkeypatch.setattr(server_module, "server_status", lambda: (False, None)) assert server_module.main(["status"]) == 0 assert os.getenv("TERRAFIN_TEST_SERVER_DOTENV") == "server-loaded" def test_env_backed_feature_lazy_loads_dotenv(tmp_path) -> None: (tmp_path / ".env").write_text("TERRAFIN_TEST_LAZY=lazy-loaded\n", encoding="utf-8") result = _run_repo_python( tmp_path, ( "import json, os; " "import TerraFin; " "from TerraFin.data.utils.api_check import check_api_key; " "print(json.dumps({" "'before': os.getenv('TERRAFIN_TEST_LAZY', ''), " "'after': check_api_key('TERRAFIN_TEST_LAZY')" "}))" ), ) assert result.stdout.strip() == '{"before": "", "after": "lazy-loaded"}' def test_configure_loads_explicit_dotenv_path_even_when_lazy_autoload_is_disabled(tmp_path) -> None: env_file = tmp_path / "custom.env" env_file.write_text("TERRAFIN_TEST_CONFIGURE=configured\n", encoding="utf-8") env = os.environ.copy() env["TERRAFIN_DISABLE_DOTENV"] = "1" result = _run_repo_python( tmp_path, ( "import os; " "from TerraFin import configure; " f"configure(dotenv_path={str(env_file).__repr__()}); " "print(os.getenv('TERRAFIN_TEST_CONFIGURE', ''))" ), env=env, ) assert result.stdout.strip() == "configured" def test_disable_dotenv_prevents_lazy_autoload(tmp_path) -> None: (tmp_path / ".env").write_text("TERRAFIN_TEST_DISABLED=should-not-load\n", encoding="utf-8") env = os.environ.copy() env["TERRAFIN_DISABLE_DOTENV"] = "1" result = _run_repo_python( tmp_path, ( "from TerraFin.data.utils.api_check import check_api_key\n" "try:\n" " check_api_key('TERRAFIN_TEST_DISABLED')\n" "except Exception as exc:\n" " print(type(exc).__name__)\n" " print(str(exc))\n" ), env=env, ) lines = result.stdout.strip().splitlines() assert lines[0] == "ValueError" assert "TERRAFIN_TEST_DISABLED" in lines[1]