cyberyan_linkedin_search/backend/tests/test_importer.py

"""The importer must normalise, enrich and stay idempotent."""

from __future__ import annotations

import textwrap
from pathlib import Path

import pytest

from apps.profiles.importers import ProfileImporter
from apps.profiles.models import Company, Education, Experience, Profile, Skill

pytestmark = pytest.mark.django_db

HEADER = (
    "full_name,first_name,last_name,gender,linkedin_url,linkedin_username,linkedin_id,"
    "industry,job_title,job_title_role,job_title_levels,job_company_name,job_company_size,"
    "location_name,location_country,inferred_years_experience,linkedin_connections,"
    "skills,experience,education,emails,phone_numbers,summary"
)

EXPERIENCE = (
    "[{'company': {'name': 'acme corp', 'size': '51-200', 'industry': 'software', "
    "'location': {'name': 'berlin, germany', 'country': 'germany'}}, "
    "'title': {'name': 'backend engineer', 'role': 'engineering', 'sub_role': 'backend', "
    "'levels': ['senior']}, 'start_date': '2019-04', 'end_date': None, 'is_primary': True}]"
)
EDUCATION = (
    "[{'school': {'name': 'tu berlin', 'type': 'post-secondary institution', "
    "'location': {'name': 'berlin, germany', 'country': 'germany'}}, "
    "'degrees': ['bachelors'], 'majors': ['computer science'], 'minors': [], "
    "'start_date': '2012', 'end_date': '2016', 'gpa': None}]"
)


def _dataset(tmp_path: Path, row: str) -> str:
    path = tmp_path / "dataset.csv"
    path.write_text(f"{HEADER}\n{row}\n", encoding="utf-8")
    return str(path)


def _row(*, with_top_level_role: bool = True) -> str:
    role = "engineering," if with_top_level_role else ""
    return (
        f"jane doe,jane,doe,female,linkedin.com/in/janedoe,janedoe,55501,software,"
        f'backend engineer,{role}"[\'senior\']",acme corp,51-200,"berlin, germany",germany,'
        f"8.0,540.0,\"['Python', 'django', 'python']\",\"{EXPERIENCE}\",\"{EDUCATION}\","
        f"\"[{{'address': 'jane@acme.com', 'type': 'professional'}}]\",\"['+491701234567']\","
        f"Backend engineer."
    )


def test_import_populates_the_relational_model(tmp_path) -> None:  # noqa: ANN001
    stats = ProfileImporter(_dataset(tmp_path, _row())).run()

    assert stats.created == 1
    profile = Profile.objects.get()
    assert profile.full_name == "jane doe"
    assert profile.years_experience == 8
    assert profile.connections == 540
    assert profile.current_company.name == "acme corp"
    assert Experience.objects.filter(profile=profile).count() == 1
    assert Education.objects.get(profile=profile).majors == ["computer science"]
    assert profile.emails == [{"address": "jane@acme.com", "type": "professional"}]


def test_skills_are_deduplicated_and_normalised(tmp_path) -> None:  # noqa: ANN001
    ProfileImporter(_dataset(tmp_path, _row())).run()
    profile = Profile.objects.get()
    names = [link.skill.name for link in profile.profile_skills.all()]
    # "Python" and "python" collapse into a single normalised skill.
    assert names == ["python", "django"]
    assert Skill.objects.get(slug="python").profile_count == 1


def test_missing_top_level_fields_are_derived_from_the_experience_blob(
    tmp_path,
) -> None:  # noqa: ANN001
    """The export frequently omits ``job_title_role``; the nested data has it."""
    ProfileImporter(_dataset(tmp_path, _row(with_top_level_role=False))).run()
    profile = Profile.objects.get()
    assert profile.job_title_role == "engineering"
    assert profile.job_title_sub_role == "backend"
    assert profile.current_company.name == "acme corp"


def test_import_is_idempotent(tmp_path) -> None:  # noqa: ANN001
    path = _dataset(tmp_path, _row())
    first = ProfileImporter(path).run()
    second = ProfileImporter(path).run()

    assert first.created == 1
    assert second.created == 0
    assert second.updated == 1
    assert Profile.objects.count() == 1
    assert Company.objects.filter(name="acme corp").count() == 1
    # Child rows are replaced, never duplicated.
    assert Experience.objects.count() == 1


def test_search_vector_is_populated(tmp_path) -> None:  # noqa: ANN001
    ProfileImporter(_dataset(tmp_path, _row())).run()
    profile = Profile.objects.get()
    assert profile.search_vector is not None
    assert "django" in profile.skills_text


def test_management_command_runs(tmp_path, capsys) -> None:  # noqa: ANN001
    from django.core.management import call_command

    call_command("import_profiles", "--path", _dataset(tmp_path, _row()), "--truncate")
    assert Profile.objects.count() == 1
    assert "profiles_created" in capsys.readouterr().out


def test_broken_dataset_reports_but_does_not_crash(tmp_path) -> None:  # noqa: ANN001
    path = tmp_path / "broken.csv"
    path.write_text(
        textwrap.dedent(
            f"""
            {HEADER}
            ,,,,,,,,,,,,,,,,,,,,,,
            """
        ).lstrip(),
        encoding="utf-8",
    )
    stats = ProfileImporter(str(path)).run()
    assert stats.report.records >= 0