test_notion_coaching_live.py
1 """Live round-trip test for coaching report export to Notion. 2 3 Creates a test page, builds coaching blocks, appends them, reads back, 4 verifies structure, sets coached checkbox, and cleans up. 5 6 Skipped by default. Run with: 7 uv run pytest tests/test_notion_coaching_live.py --run-live -v 8 9 Requires NOTION_TOKEN and NOTION_JOBS_DATABASE_ID. 10 """ 11 12 __all__: list[str] = [] 13 14 import os 15 import sys 16 from typing import Any, cast 17 18 import pytest 19 20 from config import NotionPropertyMapping 21 from integrations.notion import NotionClient 22 from models.coaching import ( 23 AnalyzedRequirement, 24 AuditedAchievement, 25 EvidenceBlock, 26 SkillAudit, 27 TailoredExperience, 28 TailoredSkills, 29 TailoredSummary, 30 ) 31 from models.notion import NotionJobPage, SyncCategory 32 from services.export_coaching_blocks import build_coaching_blocks 33 from services.sync_notion import build_page_blocks, build_page_properties 34 35 _NOTION_KEY = "NOTION_TOKEN" 36 _NOTION_DB_ID_KEY = "NOTION_JOBS_DATABASE_ID" 37 _TEST_JOB_ID = "__coaching_export_live_test__" 38 39 _skip_notion_db = pytest.mark.skipif( 40 "--run-live" not in sys.argv 41 or not os.environ.get(_NOTION_KEY) 42 or not os.environ.get(_NOTION_DB_ID_KEY), 43 reason="notion: pass --run-live and set NOTION_TOKEN + NOTION_JOBS_DATABASE_ID", 44 ) 45 46 47 def _build_test_coaching_blocks() -> tuple[list[dict[str, Any]], dict[str, int]]: 48 """Build a small but representative set of coaching blocks for testing.""" 49 analyzed = [ 50 AnalyzedRequirement( 51 text="Python 5+ years", 52 priority="MUST_HAVE", 53 priority_rationale="Core", 54 status="MATCH", 55 evidence=EvidenceBlock(skills_match=["Python"]), 56 gap_reasoning="Direct match.", 57 action="REINFORCE", 58 ), 59 AnalyzedRequirement( 60 text="Kubernetes", 61 priority="CORE", 62 priority_rationale="Infra", 63 status="MISSING", 64 evidence=EvidenceBlock(), 65 gap_reasoning="Not found.", 66 action="COVER_LETTER", 67 cover_letter_suggestion="Mention container interest.", 68 ), 69 ] 70 71 summary = TailoredSummary( 72 headline="Senior Backend Engineer", 73 summary="Tailored summary for live test.", 74 coach_rationale="Test rationale.", 75 naturalness_score=7, 76 ) 77 78 experience = TailoredExperience( 79 experience_id="exp_01", 80 company="LiveTestCorp", 81 role="Engineer", 82 kept_achievements=[ 83 AuditedAchievement( 84 text="Built scalable API", 85 decision="KEEP", 86 reason="Relevant", 87 relevance_score=9, 88 ), 89 ], 90 excluded_achievements=[ 91 AuditedAchievement( 92 text="Organized events", 93 decision="EXCLUDE", 94 reason="Irrelevant", 95 relevance_score=2, 96 ), 97 ], 98 ) 99 100 skills = TailoredSkills( 101 kept_skills=[ 102 SkillAudit( 103 skill="Python", decision="KEEP", reason="Core", relevance_score=10 104 ), 105 ], 106 excluded_skills=[], 107 required_skills=[ 108 SkillAudit( 109 skill="Kubernetes", 110 decision="REQUIRED", 111 reason="In JD", 112 relevance_score=9, 113 ), 114 ], 115 ) 116 117 result = build_coaching_blocks( 118 analyzed=analyzed, 119 tailored_summary=summary, 120 original_headline="Software Engineer", 121 original_summary="Original summary for live test.", 122 experiences=[experience], 123 skills=skills, 124 ) 125 126 return result.blocks, result.label_index 127 128 129 @_skip_notion_db 130 class TestExportCoachingRoundTrip: 131 """Live round-trip: create page -> append coaching blocks -> read back -> verify -> clean up.""" 132 133 @pytest.fixture() 134 def live_client(self) -> NotionClient: 135 return NotionClient(token=os.environ[_NOTION_KEY]) 136 137 @pytest.fixture() 138 def live_db_id(self) -> str: 139 return os.environ[_NOTION_DB_ID_KEY] 140 141 @pytest.fixture() 142 def pm(self) -> NotionPropertyMapping: 143 return NotionPropertyMapping() 144 145 def test_export_coaching_round_trip( 146 self, 147 live_client: NotionClient, 148 live_db_id: str, 149 pm: NotionPropertyMapping, 150 ) -> None: 151 """Full round-trip: create, append blocks, read back, verify, set coached, clean up.""" 152 # Clean up any leftover test pages. 153 self._cleanup(live_client, live_db_id) 154 155 page = NotionJobPage( 156 job_posting_id=_TEST_JOB_ID, 157 job_title="Coaching Export Live Test", 158 company_name="LiveTestCorp", 159 category=SyncCategory.MANUAL, 160 status="Ready to apply", 161 icon="\U0001f680", 162 ) 163 164 try: 165 # 1. Create test page. 166 page_id = live_client.create_page( 167 live_db_id, 168 build_page_properties(page, pm), 169 icon=page.icon, 170 blocks=build_page_blocks(page), 171 ) 172 assert page_id 173 174 # 2. Build coaching blocks. 175 blocks, label_index = _build_test_coaching_blocks() 176 assert len(blocks) > 0 177 178 # 3. Append blocks and capture IDs. 179 block_ids = live_client.append_blocks(page_id, blocks) 180 assert len(block_ids) == len(blocks) 181 182 # 4. Build block map. 183 block_map = { 184 label: block_ids[idx] 185 for label, idx in label_index.items() 186 if idx < len(block_ids) 187 } 188 assert "section:summary" in block_map 189 assert "section:gap" in block_map 190 assert "section:experience" in block_map 191 assert "section:skills" in block_map 192 assert "exp:exp_01" in block_map 193 assert "skill:kept" in block_map 194 assert "skill:required" in block_map 195 196 # 5. Read back page children and verify section headings. 197 children_resp = cast( 198 dict[str, Any], 199 live_client._client.blocks.children.list(block_id=page_id), 200 ) 201 children = children_resp.get("results", []) 202 heading_texts: list[str] = [] 203 for child in children: 204 if child["type"] == "heading_1": 205 heading_texts.append( 206 "".join( 207 chunk["plain_text"] 208 for chunk in child["heading_1"]["rich_text"] 209 ) 210 ) 211 assert "Summary Evolution" in heading_texts 212 assert "Gap Analysis" in heading_texts 213 assert "Experience Audit" in heading_texts 214 assert "Skills Audit" in heading_texts 215 216 # 6. Spot-check experience callout and its sibling toggle. 217 exp_block_id = block_map["exp:exp_01"] 218 exp_block = next( 219 c for c in children if c["id"] == exp_block_id 220 ) 221 assert exp_block["type"] == "callout" 222 # Achievements are sibling toggles after the callout 223 exp_idx = next( 224 i for i, c in enumerate(children) if c["id"] == exp_block_id 225 ) 226 sibling = children[exp_idx + 1] 227 assert sibling["type"] == "toggle" 228 229 # 7. Set coached checkbox. 230 live_client.update_page_property( 231 page_id, pm.coached, {"checkbox": True} 232 ) 233 234 # 8. Verify coached property. 235 page_data = cast( 236 dict[str, Any], 237 live_client._client.pages.retrieve(page_id=page_id), 238 ) 239 coached_prop = page_data["properties"][pm.coached] 240 assert coached_prop["checkbox"] is True 241 242 finally: 243 self._cleanup(live_client, live_db_id) 244 245 @staticmethod 246 def _cleanup(client: NotionClient, database_id: str) -> None: 247 """Archive any pages with the coaching export test sentinel job ID.""" 248 response = cast( 249 dict[str, Any], 250 client._client.request( 251 path=f"data_sources/{database_id}/query", 252 method="POST", 253 body={}, 254 ), 255 ) 256 pm = NotionPropertyMapping() 257 for p in response.get("results", []): 258 props = p.get("properties", {}) 259 job_id_prop = props.get(pm.job_id, {}) 260 rich_text = job_id_prop.get("rich_text", []) 261 if rich_text and rich_text[0].get("plain_text") == _TEST_JOB_ID: 262 client._client.pages.update(page_id=p["id"], archived=True)