This commit is contained in:
2025-11-14 14:47:19 +00:00
parent 2a85845835
commit 46b2319e2d
199 changed files with 607543 additions and 11147 deletions
-64
View File
@@ -1,64 +0,0 @@
from dotenv import load_dotenv, find_dotenv
load_dotenv(find_dotenv())
import os
import modules.logger_tool as logger
log_name = 'pytest_calendar'
log_dir = os.getenv("LOG_PATH", "/logs") # Default path as fallback
logging = logger.get_logger(
name=log_name,
log_level=os.getenv("LOG_LEVEL", "DEBUG"),
log_path=log_dir,
log_file=log_name,
runtime=True,
log_format='default'
)
import modules.database.tools.neo4j_driver_tools as driver_tools
import modules.database.tools.neontology_tools as neon
import pytest
from fastapi.testclient import TestClient
from routers.database.init.calendar import router
from fastapi import FastAPI
from datetime import datetime, timedelta
app = FastAPI()
app.include_router(router)
client = TestClient(app)
# Define a list of date ranges for testing
date_ranges = [
(datetime.now(), datetime.now() + timedelta(days=1)), # 1 day
(datetime.now(), datetime.now() + timedelta(days=7)), # 1 week
(datetime.now(), datetime.now() + timedelta(days=30)), # 1 month
(datetime.now(), datetime.now() + timedelta(days=183)),# 6 months
(datetime.now(), datetime.now() + timedelta(days=365)) # 1 year
]
# Fixture to manage database name increment
@pytest.fixture(scope="function", autouse=True)
def increment_db_name_counter(request):
if not hasattr(request.module, "db_name_counter"):
request.module.db_name_counter = 0
request.module.db_name_counter += 1
return request.module.db_name_counter
@pytest.mark.parametrize("start_date, end_date", date_ranges)
def test_create_calendar(start_date, end_date, increment_db_name_counter):
db_name = f"test_create_calendar_db_{increment_db_name_counter}"
neo_safe_db_name = db_name.replace("_", "")
logging.info(f"Creating calendar for {db_name} from {start_date} to {end_date}")
logging.info(f"Creating calendar for {db_name} from {start_date} to {end_date}")
response = client.post(
"/create-calendar",
params={
"db_name": neo_safe_db_name,
"start_date": start_date.strftime('%Y-%m-%d'),
"end_date": end_date.strftime('%Y-%m-%d')
}
)
assert response.status_code == 200
response_json = response.json()
assert "calendar_year_nodes" in response_json and response_json["calendar_year_nodes"] != 0
assert "calendar_month_nodes" in response_json and response_json["calendar_month_nodes"] != 0
assert "calendar_week_nodes" in response_json and response_json["calendar_week_nodes"] != 0
assert "calendar_day_nodes" in response_json and response_json["calendar_day_nodes"] != 0
-61
View File
@@ -1,61 +0,0 @@
from dotenv import load_dotenv, find_dotenv
load_dotenv(find_dotenv())
import os
import modules.logger_tool as logger
log_name = 'pytest_init_curriculum'
log_dir = os.getenv("LOG_PATH", "/logs") # Default path as fallback
logging = logger.get_logger(
name=log_name,
log_level=os.getenv("LOG_LEVEL", "DEBUG"),
log_path=log_dir,
log_file=log_name,
runtime=True,
log_format='default'
)
import modules.database.tools.neo4j_driver_tools as driver_tools
import modules.database.tools.neontology_tools as neon
import pytest
from fastapi.testclient import TestClient
from routers.database.init.curriculum import router
from fastapi import FastAPI
app = FastAPI()
app.include_router(router)
client = TestClient(app)
db_name = log_name.replace('_', '')
excel_file = os.environ['EXCEL_CURRICULUM_FILE']
driver = driver_tools.get_driver(database=db_name)
neon.init_neontology_connection()
@pytest.fixture
def sample_file():
# Use the existing Excel file to upload
file_path = excel_file
logging.info(f"Using sample file at {file_path}")
yield file_path
def test_upload_curriculum(sample_file):
db_name = "test_curriculum_db"
with open(sample_file, "rb") as f:
response = client.post(
"/upload-curriculum",
files={"file": (excel_file, f, "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet")},
data={"db_name": db_name.replace('_', '')}
)
logging.info(f"Response status code: {response.status_code}")
logging.info(f"Response JSON: {response.json()}")
assert response.status_code == 200
response_json = response.json()
logging.info(f"Response JSON keys: {response_json.keys()}")
# Adjust the assertions based on the actual response structure
assert "status" in response_json or "12" in response_json
if "status" in response_json:
assert response_json["status"] == "Success"
else:
assert "created" in response_json["12"]
assert "merged" in response_json["12"]
@@ -1,54 +0,0 @@
from dotenv import load_dotenv, find_dotenv
load_dotenv(find_dotenv())
import os
import modules.logger_tool as logger
log_name = 'pytest_timetable'
log_dir = os.getenv("LOG_PATH", "/logs") # Default path as fallback
logging = logger.get_logger(
name=log_name,
log_level=os.getenv("LOG_LEVEL", "DEBUG"),
log_path=log_dir,
log_file=log_name,
runtime=True,
log_format='default'
)
import modules.database.tools.neo4j_driver_tools as driver_tools
import modules.database.tools.neontology_tools as neon
import pytest
from fastapi.testclient import TestClient
from routers.database.init.timetable import router
from fastapi import FastAPI
import pandas as pd
app = FastAPI()
app.include_router(router)
client = TestClient(app)
db_name = log_name.replace('_', '')
excel_file = os.environ['EXCEL_TIMETABLE_FILE']
@pytest.fixture
def sample_file():
# Use the existing Excel file to upload
file_path = excel_file
logging.info(f"Using sample file at {file_path}")
yield file_path
def test_upload_school_timetable(sample_file):
db_name = "pytest_school_timetable_db"
with open(sample_file, "rb") as f:
response = client.post(
"/upload-school-timetable",
data={"db_name": db_name.replace('_', '')},
files={"file": (excel_file, f, "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet")}
)
logging.info(f"Response status code: {response.status_code}")
logging.info(f"Response JSON: {response.json()}")
assert response.status_code == 200
response_json = response.json()
assert "calendar_nodes" in response_json
assert "school_timetable_nodes" in response_json
assert response_json["calendar_nodes"] is not None
assert response_json["school_timetable_nodes"] is not None
-44
View File
@@ -1,44 +0,0 @@
from dotenv import load_dotenv, find_dotenv
load_dotenv(find_dotenv())
import os
import modules.logger_tool as logger
log_name = 'pytest_timetable'
log_dir = os.getenv("LOG_PATH", "/logs") # Default path as fallback
logging = logger.get_logger(
name=log_name,
log_level=os.getenv("LOG_LEVEL", "DEBUG"),
log_path=log_dir,
log_file=log_name,
runtime=True,
log_format='default'
)
import modules.database.tools.neo4j_driver_tools as driver_tools
import modules.database.tools.neontology_tools as neon
import pytest
from fastapi.testclient import TestClient
from fastapi import FastAPI
import pandas as pd
# Import the router from entity_init.py
from routers.database.init.entity_init import router
app = FastAPI()
app.include_router(router)
client = TestClient(app)
@pytest.mark.parametrize("username, email, user_id", [
("user1", "[email protected]", "uuid1"),
("user2", "[email protected]", "uuid2"),
("user3", "[email protected]", "uuid3")
])
def test_create_user(username, email, user_id):
response = client.post(
"/create-user",
data={"username": username, "email": email, "user_id": user_id}
)
logging.info(f"Tested creating user {username}. Response status code: {response.status_code}")
response_json = response.json()
logging.info(f"Response JSON: {response_json}")
assert response.status_code == 200
-35
View File
@@ -1,35 +0,0 @@
import os
import pytest
from fastapi.testclient import TestClient
from main import app # Adjust the import based on your project structure
client = TestClient(app)
@pytest.fixture(autouse=True)
def setup_env():
os.environ["WHISPERLIVE_HOST"] = "localhost"
os.environ["WHISPERLIVE_PORT"] = "9090"
def test_start_transcription():
user_id = "test_user"
response = client.post(f"/transcribe/live/start_transcription/{user_id}")
assert response.status_code == 200
assert response.json() == {"message": "Transcription started", "user_id": user_id}
def test_handle_whisper_live_eos_utterance():
user_id = "test_user"
data = {
"utterance": "Hello, world!",
"start": 0,
"end": 1,
"eos": True
}
response = client.post(f"/transcribe/utterance/handle_whisper_live_eos_utterance/{user_id}", json=data)
assert response.status_code == 200
assert response.json() == {"message": "Utterance logged successfully"}
def test_get_utterances():
user_id = "test_user"
response = client.get(f"/transcribe/utterance/get_utterances/{user_id}")
assert response.status_code == 200
assert "utterances" in response.json()
View File
-34
View File
@@ -1,34 +0,0 @@
from modules.whisper_live.client import TranscriptionClient
import os
import time
def setup_directories(user_dir, user_id):
user_transcript_dir = f"{user_dir}/{user_id}/transcripts"
if not os.path.exists(user_transcript_dir):
os.makedirs(user_transcript_dir)
return user_transcript_dir
def timestamped_callback(text, is_final):
if is_final:
print(f"Timestamp: {time.strftime('%H:%M:%S')}, Transcription: {text}")
def main():
user_dir = "../../data/users"
user_id = "kcar"
user_transcript_dir = setup_directories(user_dir, user_id)
client = TranscriptionClient(
"localhost",
9090,
lang="en",
translate=False,
use_vad=True,
save_output_recording=True,
output_recording_filename=f"{user_transcript_dir}/output_recording.wav",
output_transcription_path=f"{user_transcript_dir}/output.srt",
)
client()
if __name__ == "__main__":
main()
-3
View File
@@ -1,3 +0,0 @@
from modules.logger_tool import PytestFormatter
pytest_formatter = PytestFormatter()
-53
View File
@@ -1,53 +0,0 @@
def ascii_header():
return r"""
==================================================
= =
= _______ =
= | | _ =
= | |____ | | =
= | / | | | ___ __ _ _ __ =
= | | | | | / _ \/ _` | '_ \ =
= | \____| | | | __/ (_| | | | | =
= |_______| |_| \___|\__,_|_| |_| =
= =
= =
= _________ =
= | | =
= | BOOK | =
= |_________| =
= =
= =
= /\ =
= / \ =
= /____\ =
= / \ =
= / \ =
= /__________\ =
= =
= =
= _____________ =
= | | =
= | COMPUTER | =
= |_____________| =
= =
= =
= =
= _________ =
= | | =
= | TEACH | =
= |_________| =
= =
= =
= ____ =
= / \ =
= / \ =
= / \ =
= /__________\ =
= =
= =
==================================================
= =
= classroom-copilot.ai =
= =
==================================================
"""
-33
View File
@@ -1,33 +0,0 @@
import os
import requests
import pytest
import json
# Define the base URL and the tokens
base_url = f"{os.environ.get('APP_API_URL')}/arbor/data"
tokens = {
1: os.getenv("KS3_COURSE_CLASS_MEMBERSHIP_AUTH"),
2: os.getenv("TEACHING_GROUP_MEMBERSHIPS_2023_2024_AUTH"),
3: os.getenv("SCHEDULED_TIMETABLE_SLOTS_AUTH"),
4: os.getenv("BEHAVIOURAL_INCIDENTS_REPORTING_AUTH"),
5: os.getenv("Y7_LESSON_TIMETABLE_AUTH")
}
@pytest.mark.parametrize("id", [1, 2, 3, 4, 5])
def test_fetch_arbor_data(id):
token = tokens.get(id)
if not token:
pytest.fail(f"Token for ID {id} is not set")
endpoint = f"{base_url}/{id}"
headers = {"Authorization": f"Basic {token}"}
params = {"token": token}
response = requests.get(endpoint, headers=headers, params=params)
if response.status_code == 200:
print(json.dumps(response.json()))
assert response.status_code == 200
else:
pytest.fail(f"Failed for ID {id}: {response.status_code} {response.text}")
-123
View File
@@ -1,123 +0,0 @@
import os
import json
import requests
import pytest
from dotenv import load_dotenv, find_dotenv
from .formatting import ascii_header
import modules.logger_tool as logger
load_dotenv(find_dotenv())
log_name = 'api_router_graph_qa_test'
log_dir = os.getenv("LOG_PATH", "/logs") # Default path as fallback
logging = logger.get_logger(
name=log_name,
log_level=os.getenv("LOG_LEVEL", "DEBUG"),
log_path=log_dir,
log_file=log_name,
runtime=True,
log_format='default'
)
@pytest.fixture(scope="module")
def config():
return {
"database": "cc.institutes.kevlarai",
"top_k": 40,
"model": "gpt-4o",
"temperature": 0,
"verbose": False,
"return_intermediate_steps": True,
"return_direct": False,
"validate_cypher": True,
"model_type": "openai" # Default model_type
}
def load_test_cases():
with open('backend/app/tests/test_inputs/init_curriculum_db_cases.json', 'r') as f:
return json.load(f)
test_cases = load_test_cases()
@pytest.mark.parametrize("case", test_cases["curriculum_cases"])
def test_curriculum_cases(case, config):
assert run_test_case(case, config)
@pytest.mark.parametrize("case", test_cases["include_exclude_cases"]["includes"])
def test_include_cases(case, config):
assert run_test_case(case, config)
@pytest.mark.parametrize("case", test_cases["include_exclude_cases"]["excludes"])
def test_exclude_cases(case, config):
assert run_test_case(case, config)
@pytest.mark.parametrize("case", test_cases["include_exclude_cases"]["includes_excludes"])
def test_include_exclude_cases(case, config):
assert run_test_case(case, config)
def run_test_case(case, config):
logging.info(f"Starting test case with prompt: {case['prompt']}")
url = f"{os.environ['APP_API_URL']}/langchain/graph_qa/prompt"
params = {
"database": config["database"],
"prompt": case["prompt"],
"top_k": config["top_k"],
"model": config["model"],
"temperature": config["temperature"],
"verbose": config["verbose"],
"return_intermediate_steps": config["return_intermediate_steps"],
"exclude_types": case["exclude_types"],
"include_types": case["include_types"],
"return_direct": config["return_direct"],
"validate_cypher": config["validate_cypher"],
"model_type": config["model_type"]
}
logging.info(f"Constructed URL: {url}")
logging.info(f"Parameters: {params}")
try:
logging.info("Sending request to API...")
response = requests.get(url, params=params)
logging.info(f"HTTP Response Status: {response.status_code}")
response.raise_for_status()
data = response.json()
logging.info(f"Response Data: {data}")
# Log detailed test execution information
logging.info("==================================================")
logging.info("= Test Execution =")
logging.info("==================================================")
logging.info(f"= Prompt: {data.get('query', 'N/A')}")
logging.info("= =")
logging.info(f"= Query: \n{data.get('intermediate_steps', [{'query': 'N/A'}])[0].get('query', 'N/A')}")
logging.info("= =")
logging.info("==================================================")
# Determine if the test passed or failed
response_text = data.get('result', 'N/A')
context = data.get('intermediate_steps', [{'context': 'N/A'}])[1].get('context', 'N/A')
if "I don't know" in response_text or not context:
logging.error("==================================================")
logging.error("= XX Test Failed XX =")
logging.error("==================================================")
logging.error(f"= Prompt: {case['prompt']}")
logging.error(f"= Context: {context}")
logging.error(f"= Response: {response_text}")
logging.error("==================================================")
return False
else:
logging.info("==================================================")
logging.info("= ** Test Passed ** =")
logging.info("==================================================")
logging.info(f"= Prompt: {case['prompt']}")
logging.info(f"= Context: {context}")
logging.info(f"= Response: {response_text}")
logging.info("==================================================")
return True
except requests.exceptions.RequestException as e:
logging.error("==================================================")
logging.error("= ERROR =")
logging.error("==================================================")
logging.error(f"Error: {e}")
logging.error("==================================================")
return False
-294
View File
@@ -1,294 +0,0 @@
from dotenv import load_dotenv, find_dotenv
load_dotenv(find_dotenv())
import os
import pytest
from fastapi.testclient import TestClient
from fastapi import FastAPI
import json
import modules.logger_tool as logger
from routers.database.init.entity_init import router as entity_init_router
from routers.database.init.timetables import router as timetables_router
from routers.database.init.curriculum import router as curriculum_router
from backend.modules.database.schemas.entities import SchoolNode, UserNode
from modules.database.schemas.nodes.calendars import CalendarNode
# Pytest configuration
def pytest_configure(config):
config.addinivalue_line(
"markers", "school: mark test as part of school creation"
)
config.addinivalue_line(
"markers", "users: mark test as part of user creation"
)
config.addinivalue_line(
"markers", "timetable: mark test as part of timetable upload"
)
config.addinivalue_line(
"markers", "curriculum: mark test as part of curriculum upload"
)
# Setup logging
log_name = 'pytest_init_x'
log_dir = os.getenv("LOG_PATH", "/logs") # Default path as fallback
logging = logger.get_logger(
name=log_name,
log_level=os.getenv("LOG_LEVEL", "DEBUG"),
log_path=log_dir,
log_file=log_name,
runtime=True,
log_format='default'
)
# Setup FastAPI app and test client
app = FastAPI()
app.include_router(entity_init_router)
app.include_router(timetables_router)
app.include_router(curriculum_router)
client = TestClient(app)
school_timetable_file = os.environ['EXCEL_TIMETABLE_FILE']
school_curriculum_file = os.environ['EXCEL_CURRICULUM_FILE']
@pytest.fixture(scope="module")
def school_info():
db_name = "cc.institutes.devschool"
school_data = {
"db_name": db_name,
"school_uuid": "uuid1",
"school_name": "school1",
"school_website": "www.school1.com"
}
return school_data
@pytest.fixture(scope="module")
def created_school(school_info):
school_data = school_info
response = client.post("/create-school", data=school_data)
logging.info(f"Create school response: {response.json()}")
assert response.status_code == 200
logging.success("School created successfully")
response_json = response.json()
school_node = SchoolNode(**response_json["school_node"])
logging.success(f"School node created: {school_node}")
return school_node
@pytest.mark.school
def test_create_school(created_school):
school_node = created_school
assert school_node is not None
@pytest.mark.users
@pytest.mark.parametrize("user_type, expected_status", [
("standard", 200),
("developer", 200)
])
def test_create_non_school_user(user_type, expected_status):
db_name = "cc.users.devusers"
user_data = {
"user_type": user_type,
"user_name": f"test_{user_type}",
"user_email": f"test_{user_type}@example.com",
"user_id": f"{user_type}_uuid"
}
response = client.post("/create-user", data=user_data)
assert response.status_code == expected_status
logging.success(f"{user_type.capitalize()} user created successfully")
@pytest.mark.users
@pytest.mark.parametrize("user_type, expected_status", [
("cc_email_school_admin", 200),
("cc_email_teacher", 200),
("cc_email_student", 200)
])
def test_create_school_user(created_school, user_type, expected_status):
school_node = created_school
worker_data = {
"cc_email_school_admin": {
"admin_code": "ADM001",
"admin_name_formal": "Mr. Admin",
"admin_email": "[email protected]"
},
"cc_email_teacher": {
"teacher_code": "TCH001",
"teacher_name_formal": "Ms. Teacher",
"teacher_email": "[email protected]"
},
"cc_email_student": {
"student_code": "STU001",
"student_name_formal": "Student Name",
"student_email": "[email protected]"
}
}
user_data = {
"user_type": user_type,
"user_name": f"test_{user_type}",
"user_email": f"test_{user_type}@example.com",
"user_id": f"{user_type}_uuid",
"school_uuid": school_node.school_uuid,
"school_name": school_node.school_name,
"school_website": school_node.school_website,
"school_path": school_node.path,
"worker_data": json.dumps(worker_data[user_type])
}
logging.info(f"Sending user data: {user_data}")
response = client.post("/create-user", data=user_data)
assert response.status_code == expected_status
logging.success(f"{user_type.capitalize()} user created successfully")
def test_create_user_invalid_data():
invalid_user_data = {
"user_type": "invalid_type",
"user_name": "test_invalid",
"user_email": "[email protected]",
"user_id": "invalid_uuid"
}
response = client.post("/create-user", data=invalid_user_data)
assert response.status_code == 400
logging.success("Invalid user data handled correctly")
@pytest.mark.users
def test_create_school_user_without_school_node():
user_data = {
"user_type": "cc_email_teacher",
"user_name": "test_teacher_no_school",
"user_email": "[email protected]",
"user_id": "teacher_no_school_uuid"
}
response = client.post("/create-user", data=user_data)
assert response.status_code == 400
logging.success("School-related user without school_node handled correctly")
@pytest.fixture
def sample_file():
logging.info(f"Using sample file: {school_timetable_file}")
return school_timetable_file
@pytest.mark.timetable
def test_upload_school_timetable(created_school, sample_file):
school_node = created_school
with open(sample_file, "rb") as f:
response = client.post(
"/upload-school-timetable",
data={
"db_name": "cc.institutes.devschool",
"unique_id": school_node.unique_id,
"school_uuid": school_node.school_uuid,
"school_name": school_node.school_name,
"school_db_name": school_node.school_db_name,
"school_website": school_node.school_website,
"path": school_node.path
},
files={"file": (os.path.basename(sample_file), f, "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet")}
)
logging.info(f"Timetable upload response: {response.json()}")
assert response.status_code == 200
logging.success("Timetable uploaded successfully")
response_json = response.json()
for key in ["school_node", "school_calendar_nodes", "school_timetable_nodes"]:
assert key in response_json
logging.success(f"{key} present in response")
school_node = SchoolNode(**response_json["school_node"])
calendar_node = CalendarNode(**response_json['school_calendar_nodes']['calendar_node'])
logging.success(f"School node validated: {school_node}")
logging.success(f"Calendar node validated: {calendar_node}")
for key in ["school_node", "school_calendar_nodes", "school_timetable_nodes"]:
assert response_json[key] is not None
logging.success(f"{key} is not None")
logging.success("All assertions passed in test_upload_school_timetable")
@pytest.fixture
def curriculum_sample_file():
logging.info(f"Using curriculum sample file: {school_curriculum_file}")
return school_curriculum_file
@pytest.mark.curriculum
def test_upload_school_curriculum(created_school, curriculum_sample_file):
school_node = created_school
with open(curriculum_sample_file, "rb") as f:
response = client.post(
"/upload-school-curriculum",
data={
"db_name": "cc.institutes.devschool",
"school_uuid": school_node.school_uuid,
"school_name": school_node.school_name,
"school_website": school_node.school_website,
"school_path": school_node.path
},
files={"file": (os.path.basename(curriculum_sample_file), f, "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet")}
)
assert response.status_code == 200
logging.success("Curriculum uploaded successfully")
response_json = response.json()
assert "curriculum_node" in response_json
assert "pastoral_node" in response_json
assert "key_stage_nodes" in response_json
assert "year_group_syllabus_nodes" in response_json
assert "topic_nodes" in response_json
assert "topic_lesson_nodes" in response_json
assert "statement_nodes" in response_json
logging.success("All assertions passed in test_upload_school_curriculum")
@pytest.mark.users
@pytest.mark.timetable
def test_create_kcar_user_and_upload_timetable(created_school):
school_node = created_school
user_data = {
"user_type": "cc_email_teacher",
"user_name": "K Car",
"user_email": "[email protected]",
"user_id": "kcar_uuid",
"school_uuid": school_node.school_uuid,
"school_name": school_node.school_name,
"school_website": school_node.school_website,
"school_path": school_node.path,
"worker_data": json.dumps({
"teacher_code": "KCAR",
"teacher_name_formal": "Mr. K Car",
"teacher_email": "[email protected]"
})
}
logging.info(f"Creating KCar user with data: {user_data}")
response = client.post("/create-user", data=user_data)
logging.info(f"KCar user creation response: {response.json()}")
assert response.status_code == 200
logging.success("KCar user created successfully")
kcar_user = UserNode(**response.json()["data"]["user_node"])
user_timetable_file = os.environ['KCAR_TIMETABLE_URL']
logging.info(f"User timetable file: {user_timetable_file}")
with open(user_timetable_file, "rb") as f:
logging.info(f"Uploading teacher timetable for K Car: {user_timetable_file}")
response = client.post(
"/upload-worker-timetable",
data={
"user_id": kcar_user.user_id,
"db_name": "cc.institutes.devschool"
},
files={"file": (os.path.basename(user_timetable_file), f, "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet")}
)
logging.info(f"Teacher timetable upload response: {response.json()}")
assert response.status_code == 200
logging.success("K Car teacher timetable uploaded successfully")
response_json = response.json()
assert response_json["message"] == "Teacher timetable initialized successfully"
logging.success("All assertions passed in test_create_kcar_user_and_upload_timetable")
def pytest_runtest_makereport(item, call):
if call.when == "call" and call.excinfo is None:
logging.success(f"Test passed: {item.name}")
-85
View File
@@ -1,85 +0,0 @@
from dotenv import load_dotenv, find_dotenv
load_dotenv(find_dotenv())
import os
import modules.logger_tool as logger
log_name = 'api_modules_interactive_langgraph_query'
log_dir = os.getenv("LOG_PATH", "/logs") # Default path as fallback
logging = logger.get_logger(
name=log_name,
log_level=os.getenv("LOG_LEVEL", "DEBUG"),
log_path=log_dir,
log_file=log_name,
runtime=True,
log_format='default'
)
import pytest
import requests
# Define the URL of your FastAPI server
BASE_URL = "http://localhost:8000"
ENDPOINT = f"{BASE_URL}/api/langchain/interactive_langgraph_query/query"
def send_query(query):
payload = {"query": query}
headers = {"Content-Type": "application/json"}
logging.info(f"Sending query to {ENDPOINT} with payload: {payload}")
try:
response = requests.post(ENDPOINT, json=payload, headers=headers)
response.raise_for_status()
result = response.json()
logging.info(f"Received response from {ENDPOINT}: {result}")
return result
except requests.exceptions.RequestException as e:
logging.error(f"Error sending query to {ENDPOINT}: {str(e)}")
return {"error": str(e)}
@pytest.mark.simple
def test_simple_queries():
query = "Describe the relevance of Maidstone, England during the English Civil War."
logging.info(f"Running simple query test with query: {query}")
result = send_query(query)
logging.info(f"Assertion 1: Checking for absence of error")
assert "error" not in result, f"Error in response: {result.get('error')}"
logging.info(f"Assertion 2: Checking for presence of response")
assert "response" in result, "Response does not contain an answer"
logging.info(f"Assertion 3: Checking for non-empty answer")
assert len(result["response"]) > 0, "Answer is empty"
logging.info(f"All assertions passed. Response: {result['response'][:100]}...")
@pytest.mark.followup
def test_followup_queries():
initial_query = "What is the latest local news from a particular town?"
logging.info(f"Running followup query test with initial query: {initial_query}")
result = send_query(initial_query)
logging.info(f"Assertion 1: Checking for absence of error")
assert "error" not in result, f"Error in response: {result.get('error')}"
if result.get("needs_more_info", False):
logging.info("Follow-up required. Sending follow-up query.")
follow_up_query = f"{initial_query} The town is Maidstone."
follow_up_result = send_query(follow_up_query)
logging.info(f"Assertion 2: Checking for absence of error in follow-up")
assert "error" not in follow_up_result, f"Error in follow-up response: {follow_up_result.get('error')}"
logging.info(f"Assertion 3: Checking for presence of response in follow-up")
assert "response" in follow_up_result, "Follow-up response does not contain an answer"
logging.info(f"Assertion 4: Checking for non-empty answer in follow-up")
assert len(follow_up_result["response"]) > 0, "Follow-up answer is empty"
logging.info(f"All follow-up assertions passed. Response: {follow_up_result['response'][:100]}...")
else:
logging.info(f"Assertion 2: Checking for presence of response")
assert "response" in result, "Response does not contain an answer"
logging.info(f"Assertion 3: Checking for non-empty answer")
assert len(result["response"]) > 0, "Answer is empty"
logging.info(f"All assertions passed. Response: {result['response'][:100]}...")
-179
View File
@@ -1,179 +0,0 @@
from dotenv import load_dotenv, find_dotenv
load_dotenv(find_dotenv())
import os
import sys
import subprocess
from datetime import datetime
import webbrowser
import threading
import shutil
import time
# Add the parent directory to the Python path
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
import modules.logger_tool as logger
# Setup logging
log_name = 'pytest_run_tests'
log_dir = os.getenv("LOG_PATH", "/logs") # Default path as fallback
logging = logger.get_logger(
name=log_name,
log_level=os.getenv("LOG_LEVEL", "DEBUG"),
log_path=log_dir,
log_file=log_name,
runtime=True,
log_format='default'
)
def find_project_root():
# Start from the current file location
root = os.path.dirname(os.path.abspath(__file__))
# Traverse up until you find the .env file
while not os.path.exists(os.path.join(root, '.env')):
new_root = os.path.dirname(root)
if root == new_root: # root directory reached without finding .env
raise Exception("Project root not found.")
root = new_root
return root
def load_env():
project_root = find_project_root()
dotenv_path = find_dotenv(os.path.join(project_root, '.env'))
load_dotenv(dotenv_path)
required_vars = ["FIXME"]
for var in required_vars:
if var not in os.environ:
print(f"Error: {var} is not set in the environment.")
sys.exit(1)
def select_test_file():
project_root = find_project_root()
test_categories = {
"A": {
"name": "X Copilot Initialization",
"tests": {
"1": os.path.join(project_root, "backend", "app", "tests", "pytest_init_x.py")
}
},
"B": {
"name": "Graph QA",
"tests": {
"1": os.path.join(project_root, "backend", "app", "tests", "pytest_init_school_timetable_graph_qa.py"),
"2": os.path.join(project_root, "backend", "app", "tests", "pytest_init_curriculum_graph_qa.py"),
"3": os.path.join(project_root, "backend", "app", "tests", "pytest_init_calendar_graph_qa.py")
}
},
"C": {
"name": "Connections",
"tests": {
"1": os.path.join(project_root, "backend", "app", "tests", "pytest_arbor.py")
}
},
"D": {
"name": "Transcription",
"tests": {
"1": os.path.join(project_root, "tests", "pytest_transcribe.py")
}
},
"E": {
"name": "LangGraph",
"tests": {
"1": os.path.join(project_root, "backend", "app", "tests", "pytest_langgraph.py")
}
}
}
print("Select a test file to run:")
for category_key, category in test_categories.items():
print(f"\n{category_key}: {category['name']}")
for test_key, test_file in category["tests"].items():
print(f" {category_key}{test_key}: {os.path.basename(test_file)}")
choice = input("\nEnter your choice (e.g., A1): ").upper()
if len(choice) == 2 and choice[0] in test_categories and choice[1] in test_categories[choice[0]]["tests"]:
category_key, test_key = choice[0], choice[1]
return test_categories[category_key]["tests"][test_key], choice
print("Invalid choice.")
sys.exit(1)
def create_log_dir(choice, project_root):
log_dir = os.path.join(project_root, "logs", "pytests")
if choice[0] == "A":
log_dir = os.path.join(log_dir, "database", "init")
elif choice[0] == "B":
log_dir = os.path.join(log_dir, "database", "langchain", "graph_qa")
elif choice[0] == "C":
log_dir = os.path.join(log_dir, "database", "connections", "arbor")
elif choice[0] == "D":
log_dir = os.path.join(log_dir, "transcribe")
elif choice[0] == "E":
log_dir = os.path.join(log_dir, "langgraph")
else:
print("Invalid choice.")
sys.exit(1)
os.makedirs(log_dir, exist_ok=True)
return log_dir
def open_html_report_in_browser(html_path):
"""Function to open the HTML report in the default web browser."""
# Check for the existence of the file every 2 seconds, up to a maximum of 10 checks
for _ in range(10):
if os.path.exists(html_path):
webbrowser.open(html_path)
break
time.sleep(2)
else:
print("HTML report was not generated in time.")
def run_tests(test_file, log_dir, choice):
timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
base_filename = os.path.basename(test_file).replace('.py', '')
html_report = os.path.join(log_dir, f"{base_filename}_pytest_report_{timestamp}.html")
xml_report = os.path.join(log_dir, f"{base_filename}_pytest_report_{timestamp}.xml")
pytest_command = [
"pytest",
"-v",
test_file,
f"--junitxml={xml_report}",
f"--html={html_report}",
"--self-contained-html",
"--capture=tee-sys",
"--show-capture=all"
]
if choice[0] == "A":
test_components = input("Enter test components to run (school,users,timetable), comma-separated, or 'all': ").lower()
if test_components != 'all':
components = test_components.split(',')
for component in components:
pytest_command.append(f"-m {component}")
print("Running command:", ' '.join(pytest_command))
# Start a thread to open the HTML report, checking for its existence
threading.Thread(target=open_html_report_in_browser, args=(html_report,)).start()
result = subprocess.run(pytest_command, check=True)
return result
def main():
project_root = find_project_root()
load_env()
data_dir = os.path.join(project_root, "APP_DATA")
# TODO: Modify this after initial testing
if os.path.exists(data_dir):
shutil.rmtree(data_dir)
test_file, choice = select_test_file()
if not test_file:
print("Invalid choice.")
sys.exit(1)
log_dir = create_log_dir(choice, project_root)
run_tests(test_file, log_dir, choice)
if __name__ == "__main__":
main()
@@ -1,150 +0,0 @@
{
"curriculum_cases": [
{
"description": "Retrieve Information About Lessons in a Topic",
"prompt": "What are the lessons in the topic 'Maths Skills For Scientists'?",
"exclude_types": ["KeyStage", "KeyStageSyllabus", "YearGroup", "YearGroupSyllabus"],
"include_types": ["Topic", "Lesson", "LESSON_INCLUDES_LEARNING_STATEMENT"]
},
{
"description": "Retrieve Information About a Specific Year Group Syllabus",
"prompt": "What is the syllabus for Year 8?",
"exclude_types": ["KeyStage", "KeyStageSyllabus", "Topic", "Lesson", "LearningStatement"],
"include_types": ["YearGroup", "YearGroupSyllabus", "YEAR_SYLLABUS_INCLUDES_TOPIC"]
},
{
"description": "Retrieve Key Stages and Their Syllabuses",
"prompt": "What are the key stages and their syllabuses?",
"exclude_types": ["YearGroup", "YearGroupSyllabus", "Topic", "Lesson", "LearningStatement"],
"include_types": ["KeyStage", "KeyStageSyllabus", "KEY_STAGE_INCLUDES_KEY_STAGE_SYLLABUS"]
},
{
"description": "Retrieve Topics Within a Specific Year Group Syllabus",
"prompt": "What are the topics in the Year 8 Science syllabus?",
"exclude_types": ["KeyStage", "KeyStageSyllabus", "Lesson", "LearningStatement"],
"include_types": ["YearGroup", "YearGroupSyllabus", "Topic", "YEAR_SYLLABUS_INCLUDES_TOPIC"]
},
{
"description": "Retrieve All Learning Statements for a Specific Lesson",
"prompt": "What are the learning statements for the lesson '8P6.R'?",
"exclude_types": ["KeyStage", "KeyStageSyllabus", "YearGroup", "YearGroupSyllabus", "Topic"],
"include_types": ["Lesson", "LearningStatement", "LESSON_INCLUDES_LEARNING_STATEMENT"]
},
{
"description": "General Information Retrieval Without Exclusions",
"prompt": "Give me an overview of the school curriculum.",
"exclude_types": [],
"include_types": []
},
{
"description": "Retrieve Detailed Information About a Specific Node Type",
"prompt": "Give me detailed information about all topics.",
"exclude_types": [],
"include_types": ["Topic"]
},
{
"description": "Retrieve Relationships Between Specific Node Types",
"prompt": "What are the relationships between Year Groups and their syllabuses?",
"exclude_types": [],
"include_types": ["YearGroup", "YearGroupSyllabus", "KEY_STAGE_SYLLABUS_INCLUDES_YEAR_GROUP_SYLLABUS"]
}
],
"include_exclude_cases": {
"includes": [
{
"description": "Include only Lessons",
"prompt": "What are the lessons in the topic 'Maths Skills For Scientists'?",
"exclude_types": [],
"include_types": ["Lesson"]
},
{
"description": "Include only Topics",
"prompt": "What are the topics in the Year 8 Science syllabus?",
"exclude_types": [],
"include_types": ["Topic"]
},
{
"description": "Include only Year Groups",
"prompt": "What are the year groups in the school curriculum?",
"exclude_types": [],
"include_types": ["YearGroup"]
},
{
"description": "Include only Learning Statements",
"prompt": "What are the learning statements for the lesson '8P6.R'?",
"exclude_types": [],
"include_types": ["LearningStatement"]
},
{
"description": "Include only Key Stages",
"prompt": "What are the key stages in the school curriculum?",
"exclude_types": [],
"include_types": ["KeyStage"]
}
],
"excludes": [
{
"description": "Exclude Lessons",
"prompt": "What are the lessons in the topic 'Maths Skills For Scientists'?",
"exclude_types": ["Lesson"],
"include_types": []
},
{
"description": "Exclude Topics",
"prompt": "What are the topics in the Year 8 Science syllabus?",
"exclude_types": ["Topic"],
"include_types": []
},
{
"description": "Exclude Year Groups",
"prompt": "What are the year groups in the school curriculum?",
"exclude_types": ["YearGroup"],
"include_types": []
},
{
"description": "Exclude Learning Statements",
"prompt": "What are the learning statements for the lesson '8P6.R'?",
"exclude_types": ["LearningStatement"],
"include_types": []
},
{
"description": "Exclude Key Stages",
"prompt": "What are the key stages in the school curriculum?",
"exclude_types": ["KeyStage"],
"include_types": []
}
],
"includes_excludes": [
{
"description": "Include Lessons, Exclude Topics",
"prompt": "What are the lessons in the topic 'Maths Skills For Scientists'?",
"exclude_types": ["Topic"],
"include_types": ["Lesson"]
},
{
"description": "Include Topics, Exclude Lessons",
"prompt": "What are the topics in the Year 8 Science syllabus?",
"exclude_types": ["Lesson"],
"include_types": ["Topic"]
},
{
"description": "Include Year Groups, Exclude Key Stages",
"prompt": "What are the year groups in the school curriculum?",
"exclude_types": ["KeyStage"],
"include_types": ["YearGroup"]
},
{
"description": "Include Learning Statements, Exclude Lessons",
"prompt": "What are the learning statements for the lesson '8P6.R'?",
"exclude_types": ["Lesson"],
"include_types": ["LearningStatement"]
},
{
"description": "Include Key Stages, Exclude Year Groups",
"prompt": "What are the key stages in the school curriculum?",
"exclude_types": ["YearGroup"],
"include_types": ["KeyStage"]
}
]
}
}