120 lines
3.6 KiB
Python
120 lines
3.6 KiB
Python
import os
|
|
import time
|
|
from datetime import datetime, timezone
|
|
|
|
import pytest
|
|
|
|
import web.db as db
|
|
from web.utils import now_iso
|
|
|
|
|
|
# Skip this entire test module unless explicitly enabled and DB is reachable
|
|
if not os.getenv("RUN_DB_TESTS"):
|
|
pytest.skip("Set RUN_DB_TESTS=1 to run MySQL integration tests",
|
|
allow_module_level=True)
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def db_ready():
|
|
try:
|
|
db.db_init()
|
|
return True
|
|
except Exception as e:
|
|
pytest.skip(f"MySQL DB not reachable or init failed: {e}")
|
|
|
|
|
|
def unique_suffix() -> str:
|
|
# Time-based unique suffix for test records
|
|
return datetime.now(timezone.utc).strftime("%Y%m%d%H%M%S%f")
|
|
|
|
|
|
def test_user_create_and_auth(db_ready):
|
|
uname = f"it_user_{unique_suffix()}"
|
|
pw = "P@ssw0rd!"
|
|
uid = db.create_or_update_user(
|
|
uname, password=pw, is_admin=False, is_active=True)
|
|
assert isinstance(uid, int)
|
|
assert db.verify_user_credentials(uname, pw) is True
|
|
assert db.verify_user_credentials(uname, "wrong") is False
|
|
|
|
|
|
def test_regions_keywords_upsert_and_list(db_ready):
|
|
rname = f"it_region_{unique_suffix()}"
|
|
kname = f"it_keyword_{unique_suffix()}"
|
|
rid = db.upsert_region(rname)
|
|
kid = db.upsert_keyword(kname)
|
|
assert isinstance(rid, int) and rid > 0
|
|
assert isinstance(kid, int) and kid > 0
|
|
regions = db.get_all_regions()
|
|
keywords = db.get_all_keywords()
|
|
assert rname in regions
|
|
assert kname in keywords
|
|
|
|
|
|
def test_user_preferences_roundtrip(db_ready):
|
|
uname = f"it_pref_user_{unique_suffix()}"
|
|
db.create_or_update_user(uname, is_active=True)
|
|
r1, r2 = f"it_r1_{unique_suffix()}", f"it_r2_{unique_suffix()}"
|
|
k1, k2 = f"it_k1_{unique_suffix()}", f"it_k2_{unique_suffix()}"
|
|
# Set preferences (upserts regions/keywords internally if missing)
|
|
db.set_user_regions(uname, [r1, r2])
|
|
db.set_user_keywords(uname, [k1, k2])
|
|
assert set(db.get_user_regions(uname)) >= {r1, r2}
|
|
assert set(db.get_user_keywords(uname)) >= {k1, k2}
|
|
|
|
|
|
def test_upsert_listing_details_and_urls(db_ready):
|
|
# Create a unique URL
|
|
jid_suffix = unique_suffix()
|
|
url = f"https://example.org/it/{jid_suffix}.html"
|
|
db.upsert_listing(
|
|
url=url,
|
|
region="it",
|
|
keyword="integration",
|
|
title=f"IT Listing {jid_suffix}",
|
|
pay="N/A",
|
|
location="Test City",
|
|
timestamp=now_iso(),
|
|
)
|
|
job_data = {
|
|
"url": url,
|
|
"title": f"IT Job {jid_suffix}",
|
|
"company": "Acme Corp",
|
|
"location": "Test City",
|
|
"description": "A test job for integration",
|
|
"id": jid_suffix, # normalize_job_id should use this or fall back to URL
|
|
"posted_time": now_iso(),
|
|
}
|
|
db.upsert_job_details(job_data)
|
|
urls = db.db_get_all_job_urls()
|
|
assert url in urls
|
|
# Cleanup (best-effort)
|
|
try:
|
|
db.db_delete_job(jid_suffix)
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
def test_user_interactions_mark_and_visit(db_ready):
|
|
uname = f"it_user_{unique_suffix()}"
|
|
db.create_or_update_user(uname, is_active=True)
|
|
jid_suffix = unique_suffix()
|
|
url = f"https://example.org/it/{jid_suffix}.html"
|
|
db.upsert_listing(
|
|
url=url,
|
|
region="it",
|
|
keyword="interact",
|
|
title=f"IT Interact {jid_suffix}",
|
|
pay="N/A",
|
|
location="Test City",
|
|
timestamp=now_iso(),
|
|
)
|
|
# Exercise helpers — absence of exceptions is success for integration
|
|
db.mark_favorite(jid_suffix, uname, True)
|
|
db.record_visit(jid_suffix, uname, url=url)
|
|
# Cleanup
|
|
try:
|
|
db.db_delete_job(jid_suffix)
|
|
except Exception:
|
|
pass
|