2025-11-04 05:17:27 +01:00
|
|
|
import os
|
|
|
|
|
import tempfile
|
2025-11-05 15:34:23 +01:00
|
|
|
from pathlib import Path
|
2025-11-04 05:17:27 +01:00
|
|
|
from unittest.mock import MagicMock
|
|
|
|
|
|
2025-11-04 08:09:12 +01:00
|
|
|
import pytest
|
2025-11-05 15:34:23 +01:00
|
|
|
from pr.ads import AsyncDataSet
|
|
|
|
|
from pr.web.app import create_app
|
2025-11-04 08:09:12 +01:00
|
|
|
|
2025-11-04 05:17:27 +01:00
|
|
|
|
|
|
|
|
@pytest.fixture
|
|
|
|
|
def temp_dir():
|
|
|
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
|
|
|
yield tmpdir
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
@pytest.fixture
|
|
|
|
|
def mock_api_response():
|
|
|
|
|
return {
|
2025-11-04 08:09:12 +01:00
|
|
|
"choices": [{"message": {"role": "assistant", "content": "Test response"}}],
|
|
|
|
|
"usage": {"prompt_tokens": 10, "completion_tokens": 5, "total_tokens": 15},
|
2025-11-04 05:17:27 +01:00
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
@pytest.fixture
|
|
|
|
|
def mock_args():
|
|
|
|
|
args = MagicMock()
|
|
|
|
|
args.message = None
|
|
|
|
|
args.model = None
|
|
|
|
|
args.api_url = None
|
|
|
|
|
args.model_list_url = None
|
|
|
|
|
args.interactive = False
|
|
|
|
|
args.verbose = False
|
|
|
|
|
args.no_syntax = False
|
|
|
|
|
args.include_env = False
|
|
|
|
|
args.context = None
|
|
|
|
|
args.api_mode = False
|
|
|
|
|
return args
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
@pytest.fixture
|
|
|
|
|
def sample_context_file(temp_dir):
|
2025-11-04 08:09:12 +01:00
|
|
|
context_path = os.path.join(temp_dir, ".rcontext.txt")
|
|
|
|
|
with open(context_path, "w") as f:
|
|
|
|
|
f.write("Sample context content\n")
|
2025-11-04 05:17:27 +01:00
|
|
|
return context_path
|
2025-11-05 15:34:23 +01:00
|
|
|
|
|
|
|
|
|
|
|
|
|
@pytest.fixture
|
|
|
|
|
async def client(aiohttp_client, monkeypatch):
|
|
|
|
|
"""Create a test client for the app."""
|
|
|
|
|
with tempfile.NamedTemporaryFile(delete=False, suffix=".db") as tmp:
|
|
|
|
|
temp_db_file = tmp.name
|
|
|
|
|
|
|
|
|
|
# Monkeypatch the db
|
|
|
|
|
monkeypatch.setattr("pr.web.views.base.db", AsyncDataSet(temp_db_file, f"{temp_db_file}.sock"))
|
|
|
|
|
|
|
|
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
|
|
|
tmp_path = Path(tmpdir)
|
|
|
|
|
static_dir = tmp_path / "static"
|
|
|
|
|
templates_dir = tmp_path / "templates"
|
|
|
|
|
repos_dir = tmp_path / "repos"
|
|
|
|
|
|
|
|
|
|
static_dir.mkdir()
|
|
|
|
|
templates_dir.mkdir()
|
|
|
|
|
repos_dir.mkdir()
|
|
|
|
|
|
|
|
|
|
# Monkeypatch the directories
|
|
|
|
|
monkeypatch.setattr("pr.web.config.STATIC_DIR", static_dir)
|
|
|
|
|
monkeypatch.setattr("pr.web.config.TEMPLATES_DIR", templates_dir)
|
|
|
|
|
monkeypatch.setattr("pr.web.config.REPOS_DIR", repos_dir)
|
|
|
|
|
monkeypatch.setattr("pr.web.config.REPOS_DIR", repos_dir)
|
|
|
|
|
|
|
|
|
|
# Create minimal templates
|
|
|
|
|
(templates_dir / "index.html").write_text("<html>Index</html>")
|
|
|
|
|
(templates_dir / "login.html").write_text("<html>Login</html>")
|
|
|
|
|
(templates_dir / "register.html").write_text("<html>Register</html>")
|
|
|
|
|
(templates_dir / "dashboard.html").write_text("<html>Dashboard</html>")
|
|
|
|
|
(templates_dir / "repos.html").write_text("<html>Repos</html>")
|
|
|
|
|
(templates_dir / "api_keys.html").write_text("<html>API Keys</html>")
|
|
|
|
|
(templates_dir / "repo.html").write_text("<html>Repo</html>")
|
|
|
|
|
(templates_dir / "file.html").write_text("<html>File</html>")
|
|
|
|
|
(templates_dir / "edit_file.html").write_text("<html>Edit File</html>")
|
|
|
|
|
(templates_dir / "deploy.html").write_text("<html>Deploy</html>")
|
|
|
|
|
|
|
|
|
|
app_instance = await create_app()
|
|
|
|
|
client = await aiohttp_client(app_instance)
|
|
|
|
|
|
|
|
|
|
yield client
|
|
|
|
|
|
|
|
|
|
# Cleanup db
|
|
|
|
|
os.unlink(temp_db_file)
|
|
|
|
|
sock_file = f"{temp_db_file}.sock"
|
|
|
|
|
if os.path.exists(sock_file):
|
|
|
|
|
os.unlink(sock_file)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
@pytest.fixture
|
|
|
|
|
async def authenticated_client(client):
|
|
|
|
|
"""Create a client with an authenticated user."""
|
|
|
|
|
# Register a user
|
|
|
|
|
resp = await client.post(
|
|
|
|
|
"/register",
|
|
|
|
|
data={
|
|
|
|
|
"username": "testuser",
|
|
|
|
|
"email": "test@example.com",
|
|
|
|
|
"password": "password123",
|
|
|
|
|
"confirm_password": "password123",
|
|
|
|
|
},
|
|
|
|
|
allow_redirects=False,
|
|
|
|
|
)
|
|
|
|
|
assert resp.status == 302 # Redirect to login
|
|
|
|
|
|
|
|
|
|
# Login
|
|
|
|
|
resp = await client.post(
|
|
|
|
|
"/login", data={"username": "testuser", "password": "password123"}, allow_redirects=False
|
|
|
|
|
)
|
|
|
|
|
assert resp.status == 302 # Redirect to dashboard
|
|
|
|
|
|
|
|
|
|
return client
|
