123 lines
4.6 KiB
Python
123 lines
4.6 KiB
Python
import asyncio
|
|
import tempfile
|
|
import os
|
|
import shutil
|
|
from unittest.mock import AsyncMock, patch, MagicMock
|
|
from db.models import TargetChannel, AIProviderProfile
|
|
from core.llm import LLMClient
|
|
from services.ai_processor import AIProcessor
|
|
from core.metrics import (
|
|
DISK_TOTAL_BYTES,
|
|
DISK_USED_BYTES,
|
|
DISK_FREE_BYTES,
|
|
DISK_FREE_PERCENT,
|
|
update_disk_metrics
|
|
)
|
|
from services.metrics_reporter import get_instant_metrics_report
|
|
|
|
|
|
def test_disk_metrics_gauge():
|
|
update_disk_metrics()
|
|
sample = DISK_TOTAL_BYTES.collect()[0].samples
|
|
assert len(sample) > 0
|
|
for s in sample:
|
|
assert s.value > 0
|
|
|
|
|
|
async def test_metrics_report_includes_disk():
|
|
fake_vector_free = [
|
|
{"metric": {"mountpoint": "/"}, "value": [0, str(20.0 * 1024 ** 3)]},
|
|
{"metric": {"mountpoint": "/projects"}, "value": [0, str(35.0 * 1024 ** 3)]}
|
|
]
|
|
fake_vector_total = [
|
|
{"metric": {"mountpoint": "/"}, "value": [0, str(200.0 * 1024 ** 3)]},
|
|
{"metric": {"mountpoint": "/projects"}, "value": [0, str(40.0 * 1024 ** 3)]}
|
|
]
|
|
fake_vector_pct = [
|
|
{"metric": {"mountpoint": "/"}, "value": [0, "10.0"]},
|
|
{"metric": {"mountpoint": "/projects"}, "value": [0, "87.5"]}
|
|
]
|
|
|
|
async def fake_query_vector(query):
|
|
if "copykar_disk_free_bytes" in query:
|
|
return fake_vector_free
|
|
if "copykar_disk_total_bytes" in query:
|
|
return fake_vector_total
|
|
if "copykar_disk_free_percent" in query:
|
|
return fake_vector_pct
|
|
return []
|
|
|
|
with patch("services.metrics_reporter._query_instant", return_value=0.0), \
|
|
patch("services.metrics_reporter._query_vector", side_effect=fake_query_vector):
|
|
|
|
report = await get_instant_metrics_report()
|
|
assert "فضای ذخیرهسازی تفکیکی درایوها (Mount Points Storage)" in report
|
|
assert "<code>/</code>" in report
|
|
assert "<code>/projects</code>" in report
|
|
assert "20.00 GB" in report
|
|
assert "35.00 GB" in report
|
|
assert "10.0%" in report
|
|
assert "87.5%" in report
|
|
|
|
|
|
async def test_multimodal_vision_image_payload():
|
|
with tempfile.NamedTemporaryFile(suffix=".jpg", delete=False) as tmp:
|
|
tmp.write(b"fake-image-binary-data")
|
|
tmp_path = tmp.name
|
|
|
|
try:
|
|
profile_openai = AIProviderProfile(
|
|
id=1,
|
|
name="OpenAI Vision",
|
|
provider_type="openai",
|
|
model="gpt-4o",
|
|
is_active=True
|
|
)
|
|
repo_mock = AsyncMock()
|
|
repo_mock.get_active_provider_profile.return_value = profile_openai
|
|
repo_mock.get_provider_profiles.return_value = [profile_openai]
|
|
repo_mock.get_setting.return_value = "false"
|
|
repo_mock.record_ai_log = AsyncMock()
|
|
|
|
client = LLMClient(repo=repo_mock)
|
|
|
|
# 1. Test OpenAI vision call formatting
|
|
captured_messages = []
|
|
async def fake_post(url, headers=None, json=None):
|
|
captured_messages.extend(json.get("messages", []))
|
|
mock_resp = MagicMock()
|
|
mock_resp.raise_for_status = MagicMock()
|
|
mock_resp.json.return_value = {
|
|
"choices": [{"message": {"content": '{"decision": "accept", "rewritten_text": "Image saw a cat"}'}}]
|
|
}
|
|
return mock_resp
|
|
|
|
with patch("httpx.AsyncClient.post", side_effect=fake_post):
|
|
target = TargetChannel(id=1, channel_id=-100123456, title="Vision Channel", username="vision_ch", language="fa", personality="طنز")
|
|
processor = AIProcessor(repo=repo_mock, llm=client)
|
|
res = await processor.rewrite_for_target("عکس را ببین", target, has_media=True, image_path=tmp_path)
|
|
assert res.is_rejected is False
|
|
assert "Image saw a cat" in str(res)
|
|
|
|
user_msg = [m for m in captured_messages if m.get("role") == "user"][0]
|
|
assert isinstance(user_msg["content"], list)
|
|
types = [part["type"] for part in user_msg["content"]]
|
|
assert "text" in types
|
|
assert "image_url" in types
|
|
img_url = [part["image_url"]["url"] for part in user_msg["content"] if part["type"] == "image_url"][0]
|
|
assert img_url.startswith("data:image/jpeg;base64,")
|
|
|
|
finally:
|
|
if os.path.exists(tmp_path):
|
|
os.remove(tmp_path)
|
|
|
|
|
|
async def main():
|
|
test_disk_metrics_gauge()
|
|
await test_metrics_report_includes_disk()
|
|
await test_multimodal_vision_image_payload()
|
|
print("All disk metrics and multimodal vision image passing tests passed successfully!")
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|