"""Unit tests for talent/plugins.py — the pure functions only. No HTTP-call tests here, matching the Buffer adapter's precedent: the request helpers are thin httpx wrappers and the live smoke run covers them. """ from __future__ import annotations from talent import plugins # ---------------------------------------------------------------- build_actor_input def test_title_goes_to_the_facet_and_skills_to_the_query(): result = plugins.build_actor_input( { "title": "Backend Engineer", "requirements": ["Python", "FastAPI", "PostgreSQL", "Docker", "AWS"], "location": "Berlin", }, max_results=10, ) assert result["currentJobTitles"] == ["Backend Engineer"] assert result["searchQuery"] == "Python FastAPI PostgreSQL" assert result["maxItems"] == 10 assert result["locations"] == ["Berlin"] assert result["profileScraperMode"] == plugins.APIFY_PROFILE_MODE def test_actor_input_omits_locations_and_query_when_job_has_none(): result = plugins.build_actor_input({"title": "Designer", "requirements": []}, max_results=5) assert "locations" not in result assert "searchQuery" not in result # the title facet alone carries the search assert result["currentJobTitles"] == ["Designer"] def test_non_geographic_locations_are_not_sent_as_filters(): for value in ("Remote", "remote", "HYBRID", "Work From Home", " Onsite "): result = plugins.build_actor_input( {"title": "Dev", "requirements": [], "location": value}, max_results=5 ) assert "locations" not in result, value def test_remote_override_clears_the_location_filter_entirely(): # An explicit "Remote" override means "no geography constraint" — it must # not be sent as a filter AND must not fall back to the job's location. result = plugins.build_actor_input( {"title": "Dev", "requirements": [], "location": "Berlin"}, max_results=5, overrides={"location": "Remote"}, ) assert "locations" not in result def test_facet_title_is_capped_at_100_chars(): result = plugins.build_actor_input( {"title": "X" * 300, "requirements": []}, max_results=5 ) assert result["currentJobTitles"] == ["X" * 100] def test_actor_input_overrides_win(): result = plugins.build_actor_input( {"title": "Backend Engineer", "requirements": ["Python"], "location": "Berlin"}, max_results=5, overrides={"keywords": "data engineer spark", "location": "Munich"}, ) assert result["searchQuery"] == "data engineer spark" assert result["locations"] == ["Munich"] assert result["currentJobTitles"] == ["Backend Engineer"] def test_actor_input_ignores_blank_requirement_entries(): result = plugins.build_actor_input( {"title": "Dev", "requirements": [" ", "", "Go"]}, max_results=5 ) assert result["searchQuery"] == "Go" def test_sentence_requirements_stay_out_of_the_query(): result = plugins.build_actor_input( { "title": "Amazon PPC", "requirements": [ "2-5 years of experience managing Amazon PPC campaigns for e-commerce brands", "Strong hands-on experience with Amazon Ads, including Sponsored Products", ], }, max_results=5, ) assert "searchQuery" not in result assert result["currentJobTitles"] == ["Amazon PPC"] def test_optional_skills_fill_in_when_requirements_are_prose(): result = plugins.build_actor_input( { "title": "Amazon PPC", "requirements": ["Several sentences of prose describing years of experience required"], "optional_skills": ["Amazon Seller Central", "Helium 10", "PPC Bid Management", "Extra"], }, max_results=5, ) assert result["searchQuery"] == "Amazon Seller Central Helium 10 PPC Bid Management" def test_keyword_requirements_win_over_optional_skills(): result = plugins.build_actor_input( { "title": "Dev", "requirements": ["Python", "FastAPI"], "optional_skills": ["Docker", "AWS"], }, max_results=5, ) assert result["searchQuery"] == "Python FastAPI Docker" def test_experience_range_selects_overlapping_buckets(): assert plugins.years_of_experience_ids(3, 5) == ["3"] assert plugins.years_of_experience_ids(2, 4) == ["2", "3"] assert plugins.years_of_experience_ids(5, None) == ["3", "4", "5"] assert plugins.years_of_experience_ids(None, 1) == ["1", "2"] assert plugins.years_of_experience_ids(0, 60) == ["1", "2", "3", "4", "5"] assert plugins.years_of_experience_ids(None, None) == [] def test_actor_input_carries_experience_filter(): result = plugins.build_actor_input( {"title": "Dev", "requirements": [], "experience_min": 3, "experience_max": 5}, max_results=5, ) assert result["yearsOfExperienceIds"] == ["3"] no_exp = plugins.build_actor_input({"title": "Dev", "requirements": []}, max_results=5) assert "yearsOfExperienceIds" not in no_exp def test_actor_input_start_page(): paged = plugins.build_actor_input({"title": "Dev"}, max_results=5, start_page=3) assert paged["startPage"] == 3 first = plugins.build_actor_input({"title": "Dev"}, max_results=5, start_page=1) assert "startPage" not in first # page 1 is the actor default; keep input stable capped = plugins.build_actor_input({"title": "Dev"}, max_results=5, start_page=999) assert capped["startPage"] == 100 def test_actor_input_excludes_own_company_urls(): result = plugins.build_actor_input({"title": "Dev"}, max_results=5) assert result["excludeCurrentCompanies"] == plugins.APIFY_EXCLUDE_COMPANY_URLS assert any("utopiadeals" in u for u in result["excludeCurrentCompanies"]) # ---------------------------------------------------------------- own-company filter def test_current_utopia_employees_are_excluded(): assert plugins.is_excluded_profile({"current_company": "Utopia Brands"}) assert plugins.is_excluded_profile({"current_company": "utopia deals"}) assert plugins.is_excluded_profile({"current_company": "Utopia Brands Pakistan (Pvt) Ltd"}) def test_other_companies_and_former_employees_pass(): assert not plugins.is_excluded_profile({"current_company": "Acme"}) assert not plugins.is_excluded_profile({"current_company": None}) assert not plugins.is_excluded_profile({}) # Headline mentioning Utopia does NOT exclude someone whose current company # is elsewhere (e.g. "ex-Utopia Deals, now at Acme"). assert not plugins.is_excluded_profile( {"current_company": "Acme", "headline": "ex-Utopia Deals engineer"} ) def test_headline_is_the_fallback_when_company_is_missing(): assert plugins.is_excluded_profile( {"current_company": None, "headline": "SEO Executive at Utopia Deals"} ) assert not plugins.is_excluded_profile( {"current_company": None, "headline": "Backend Engineer"} ) # ---------------------------------------------------------------- local_status def test_every_known_apify_status_maps(): assert plugins.local_status("READY") == "running" assert plugins.local_status("RUNNING") == "running" assert plugins.local_status("TIMING-OUT") == "running" assert plugins.local_status("ABORTING") == "running" assert plugins.local_status("SUCCEEDED") == "succeeded" assert plugins.local_status("FAILED") == "failed" assert plugins.local_status("TIMED-OUT") == "timed_out" assert plugins.local_status("ABORTED") == "aborted" def test_unknown_and_missing_statuses_stay_running(): assert plugins.local_status("SOMETHING-NEW") == "running" assert plugins.local_status(None) == "running" assert plugins.local_status("") == "running" def test_terminal_statuses_are_the_terminal_local_values(): assert plugins.TERMINAL_STATUSES == {"succeeded", "failed", "timed_out", "aborted"} # ---------------------------------------------------------------- normalize_linkedin_url def test_url_normalization_canonicalizes(): expected = "https://www.linkedin.com/in/jane-doe" assert plugins.normalize_linkedin_url("https://www.linkedin.com/in/Jane-Doe/") == expected assert plugins.normalize_linkedin_url("http://www.LinkedIn.com/in/jane-doe?src=x#top") == expected assert plugins.normalize_linkedin_url("www.linkedin.com/in/jane-doe") == expected def test_url_normalization_rejects_non_linkedin(): assert plugins.normalize_linkedin_url("https://twitter.com/janedoe") is None assert plugins.normalize_linkedin_url("") is None assert plugins.normalize_linkedin_url(None) is None # ---------------------------------------------------------------- normalize_profile # Shape observed live from harvestapi~linkedin-profile-search (Full mode). RICH_ITEM = { "linkedinUrl": "https://www.linkedin.com/in/sachinsharma31261", "publicIdentifier": "sachinsharma31261", "firstName": "Sachin", "lastName": "Sharma", "headline": "Software Engineer @ Lucid Motors", "about": "Staff Software Engineer with 10 years of experience.", "location": {"linkedinText": "San Jose, California, United States"}, "photo": "https://media.licdn.com/photo.jpg", "currentPosition": {"title": "Lead Software Engineer", "companyName": "Lucid Motors"}, "experience": [{"title": "Lead Software Engineer", "companyName": "Lucid Motors"}], "skills": [{"name": "Java"}, {"name": "Python"}, {"name": "Java"}], } def test_rich_item_normalizes_every_card_field(): profile = plugins.normalize_profile(RICH_ITEM) assert profile["linkedin_url"] == "https://www.linkedin.com/in/sachinsharma31261" assert profile["public_id"] == "sachinsharma31261" assert profile["full_name"] == "Sachin Sharma" assert profile["headline"] == "Software Engineer @ Lucid Motors" assert profile["location"] == "San Jose, California, United States" assert profile["current_title"] == "Lead Software Engineer" assert profile["current_company"] == "Lucid Motors" assert profile["avatar_url"] == "https://media.licdn.com/photo.jpg" assert profile["summary"] == "Staff Software Engineer with 10 years of experience." assert profile["skills"] == ["Java", "Python"] # dict entries, deduped assert profile["raw"] is RICH_ITEM def test_skills_accept_plain_strings_and_prefer_top_skills(): profile = plugins.normalize_profile({ "linkedinUrl": "https://linkedin.com/in/x", "topSkills": ["Go", "Rust"], "skills": [{"name": "Ignored"}], }) assert profile["skills"] == ["Go", "Rust"] none = plugins.normalize_profile({"linkedinUrl": "https://linkedin.com/in/y"}) assert none["skills"] == [] def test_minimal_item_still_normalizes(): profile = plugins.normalize_profile( {"url": "https://linkedin.com/in/someone", "name": "Some One"} ) assert profile["linkedin_url"] == "https://linkedin.com/in/someone" assert profile["full_name"] == "Some One" assert profile["headline"] is None assert profile["avatar_url"] is None def test_item_without_linkedin_url_is_skipped_not_fatal(): assert plugins.normalize_profile({"name": "No Url"}) is None assert plugins.normalize_profile({"url": "https://example.com/x"}) is None assert plugins.normalize_profile("not a dict") is None def test_location_accepts_plain_string(): profile = plugins.normalize_profile( {"linkedinUrl": "https://linkedin.com/in/x", "location": "Greater St. Louis"} ) assert profile["location"] == "Greater St. Louis" # ---------------------------------------------------------------- broadening ladder def test_broadening_ladder_relaxes_one_constraint_per_rung(): original = { "currentJobTitles": ["Amazon PPC"], "searchQuery": "Amazon DSP experience", "locations": ["Karachi, Pakistan"], "yearsOfExperienceIds": ["2", "3"], "maxItems": 25, "profileScraperMode": "Full", } rung1 = plugins.broaden_actor_input(original) assert "searchQuery" not in rung1 assert rung1["currentJobTitles"] == ["Amazon PPC"] assert rung1["locations"] == ["Karachi, Pakistan"] # never relaxed assert rung1["yearsOfExperienceIds"] == ["2", "3"] # never relaxed rung2 = plugins.broaden_actor_input(rung1) assert "currentJobTitles" not in rung2 assert rung2["searchQuery"] == "Amazon PPC" # title as keywords assert rung2["locations"] == ["Karachi, Pakistan"] assert plugins.broaden_actor_input(rung2) is None # exhausted def test_broadening_does_not_mutate_the_original_input(): original = {"currentJobTitles": ["Dev"], "searchQuery": "Python"} plugins.broaden_actor_input(original) assert original == {"currentJobTitles": ["Dev"], "searchQuery": "Python"} # ---------------------------------------------------------------- relevance score AI_JOB = { "title": "AI Engineer", "requirements": ["Python", "FastAPI", "PostgreSQL"], "optional_skills": [], } def test_actual_ai_engineer_outranks_keyword_stuffed_full_stack(): # The live case that motivated the phrase rule: Khawar's keyword-stuffed # headline carries every hot token ("AI/ML Engineer | Python | FastAPI | # ...") but his title is Full Stack; Shaheer's title IS "AI Engineer". full_stack = { "current_title": "Sr. Full Stack Engineer", "headline": ( "Senior Software Engineer| Senior Full Stack Engineer | AI/ML Engineer " "| Python | FastAPI | Django | React| LLMs | RAG | Agentic AI | AWS" ), "skills": ["Python (Programming Language)", "JavaScript", "React.js"], "summary": "Senior Software Engineer delivering web applications with PostgreSQL.", } ai_engineer = { "current_title": "AI Engineer", "headline": "AI Engineer @ EmpireOne | Building Production LLM Systems", "skills": ["Keras", "Docker", "FastAPI", "PostgreSQL", "Python"], "summary": "Machine Learning and Data Science.", } weak = plugins.relevance_score(AI_JOB, full_stack) strong = plugins.relevance_score(AI_JOB, ai_engineer) assert strong > weak assert strong >= 55 # exact title phrase at minimum assert weak <= 80 # scattered tokens cap at 35 + full skills 45 def test_relevance_score_bounds_and_empty_profile(): perfect = plugins.relevance_score(AI_JOB, { "current_title": "AI Engineer", "skills": ["Python", "FastAPI", "PostgreSQL"], }) assert perfect == 100 assert plugins.relevance_score(AI_JOB, {}) == 0 assert plugins.relevance_score({"title": "", "requirements": []}, {"headline": "x"}) == 0 def test_skills_overlap_is_graded_not_all_or_nothing(): # A single unmatched niche term must not zero the whole skills component: # that put an entire live pool on exactly 60. job = {"title": "PPC", "requirements": ["Amazon Seller Central"], "optional_skills": []} full = plugins.relevance_score(job, {"skills": ["Amazon Seller Central"], "current_title": "PPC"}) partial = plugins.relevance_score(job, {"skills": ["Amazon"], "current_title": "PPC"}) none = plugins.relevance_score(job, {"skills": ["Photoshop"], "current_title": "PPC"}) assert full == 100 assert none == 55 # title only assert none < partial < full # 1 of 3 tokens matched sits in between def test_prose_requirements_still_differentiate_profiles(): # The Amazon PPC case: prose requirements yielded one niche term and every # sourced profile scored identically. Graded token overlap must spread them. job = { "title": "Amazon PPC", "requirements": [ "2-5 years of experience managing Amazon PPC campaigns for e-commerce brands", "Strong hands-on experience with Amazon Ads, including Sponsored Products", ], "optional_skills": [], } rich = plugins.relevance_score(job, { "current_title": "Amazon PPC Manager", "skills": ["Amazon PPC", "PPC Bid Management", "Amazon Listing Optimization"], "summary": "Managing Amazon Ads campaigns, Sponsored Products and Sponsored Display for e-commerce brands.", }) thin = plugins.relevance_score(job, { "current_title": "Amazon PPC Specialist", "skills": [], "summary": "", }) assert rich > thin >= 55 assert rich - thin >= 15 # a real spread, not a flat pool def test_headline_phrase_scores_below_title_phrase(): job = {"title": "AI Engineer", "requirements": [], "optional_skills": []} in_title = plugins.relevance_score(job, {"current_title": "AI Engineer"}) in_headline = plugins.relevance_score(job, {"current_title": "Developer", "headline": "AI Engineer at Acme"}) scattered = plugins.relevance_score(job, {"current_title": "Engineer", "headline": "Agentic AI | Python"}) assert in_title == 55 assert in_headline == 45 assert scattered == 35 # tokens split across title and headline never combine def test_title_containment_scores_like_an_exact_title(): # Live case: job "Generative Engineer", pool titled "Generative AI # Engineer" — the exact phrase never occurs, so every genuine match fell # to the scattered 35 tier and the whole pool compressed into the 40s. job = {"title": "Generative Engineer", "requirements": [], "optional_skills": []} interleaved = plugins.relevance_score(job, {"current_title": "Generative AI Engineer"}) senior = plugins.relevance_score(job, {"current_title": "Senior Generative AI Engineer"}) assert interleaved == 55 assert senior == 55 # Containment applies to the TITLE only: the same tokens scattered across # a keyword-stuffed headline still cap at the 35 tier. stuffed = plugins.relevance_score( job, {"current_title": "Developer", "headline": "Generative AI | Engineer | Python"}, ) assert stuffed == 35 # ---------------------------------------------------------------- detail extraction RAW_WITH_HISTORY = { "experience": [ { "position": "Freelance", "companyName": "Upwork", "employmentType": "Self-employed", "location": "Rawalpindi, Punjab, Pakistan", "duration": "1 yr 3 mos", "description": None, "skills": ["Amazon Seller Central", "Amazon PPC"], "startDate": {"month": "Jun", "year": 2025, "text": "Jun 2025"}, "endDate": {"text": "Present"}, }, "not a dict", ], "education": [ { "schoolName": "Modern Public School - Pakistan", "degree": "Intermediate", "fieldOfStudy": "Computer Science", "period": "May 2020 - Jun 2022", }, ], } def test_experience_extraction_matches_live_shape(): entries = plugins.extract_experience(RAW_WITH_HISTORY) assert len(entries) == 1 entry = entries[0] assert entry["title"] == "Freelance" assert entry["company"] == "Upwork" assert entry["employment_type"] == "Self-employed" assert entry["duration"] == "1 yr 3 mos" assert entry["period"] == "Jun 2025 – Present" assert entry["description"] is None assert entry["skills"] == ["Amazon Seller Central", "Amazon PPC"] def test_education_extraction_matches_live_shape(): entries = plugins.extract_education(RAW_WITH_HISTORY) assert entries == [{ "school": "Modern Public School - Pakistan", "degree": "Intermediate", "field": "Computer Science", "period": "May 2020 - Jun 2022", }] def test_history_extraction_tolerates_empty_raw(): assert plugins.extract_experience({}) == [] assert plugins.extract_education({}) == [] assert plugins.extract_experience(None) == [] assert plugins.extract_education(None) == [] def test_company_from_nested_experience_company_dict(): profile = plugins.normalize_profile({ "linkedinUrl": "https://linkedin.com/in/x", "experience": [{"title": "Engineer", "company": {"name": "Acme"}}], }) assert profile["current_title"] == "Engineer" assert profile["current_company"] == "Acme"