19721176345

Committed 27 Nov 2025 12:20AM UTC coverage: 73.564% (+5.0%) from 68.573%

Build # 19721176345

Build Type

push

github

Committed by

web-flow

Commit Message

refactor: extract access helpers and services from http_handler.py (#17)

Run Details

224 of 273 new or added lines in 6 files covered. (82.05%)

11 existing lines in 1 file now uncovered.

1486 of 2020 relevant lines covered (73.56%)

0.74 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

0.0

/src/drive_audit/http_handler.py

1	"""HTTP handler for managing Google Drive access."""
2
UNCOV 3	import json	×
UNCOV 4	import logging	×
NEW 5	from typing import Any, Dict, List	×
6
NEW 7	from .access_service import (	×
8	create_client_folder,
9	extract_folder_id,
10	get_task_and_assignees,
11	grant_access,
12	normalize_assignee_ids,
13	parse_assignee_ids,
14	)
NEW 15	from .http_utils import JsonRequestHandler, LocalizedError	×
UNCOV 16	from .model import DriveConfig, HttpConfig	×
UNCOV 17	from .planfix_client import PlanfixClient	×
UNCOV 18	from .translations import translate	×
19
UNCOV 20	logger = logging.getLogger(__name__)	×
21
22
UNCOV 23	def create_handler(	×
24	planfix_client: PlanfixClient,
25	service,
26	http_config: HttpConfig,
27	drive_config: DriveConfig,
28	role: str,
29	):
30	language = http_config.lang	×
31
NEW 32	class AccessHandler(JsonRequestHandler):	×
NEW 33	http_config = http_config	×
NEW 34	language = language	×
35
36	def _log_request(self, payload: Dict[str, Any]) -> None:	×
37	logger.info(	×
38	"%s request: %s", self.path, json.dumps(payload, ensure_ascii=False)
39	)
40
41	def _format_accounts(self, accounts: List[str]) -> str:	×
NEW 42	return ", ".join(accounts) if accounts else self.translate("none")	×
43
44	def _handle_set_client_folder_access(self, payload: Dict[str, Any]) -> None:	×
45	required_fields = ["contact_id", "folder_url"]	×
46	missing_fields = [	×
47	field for field in required_fields if field not in payload
48	]
49	if missing_fields:	×
NEW 50	self.send_json(	×
51	200,
52	{
53	"answer": self.translate(
54	"missing_fields", fields=", ".join(missing_fields)
55	)
56	},
57	)
58	return	×
59
60	try:	×
61	contact_id = int(payload["contact_id"])	×
62	folder_id = extract_folder_id(str(payload["folder_url"]))	×
63
64	has_task_id = "task_id" in payload	×
65	has_assignee_id = "assignee_id" in payload	×
66
67	if has_task_id and has_assignee_id:	×
68	task_id = int(payload["task_id"])	×
69	initial_assignee_ids = normalize_assignee_ids(	×
70	parse_assignee_ids(payload["assignee_id"])
71	)
72	elif not has_task_id and not has_assignee_id:	×
NEW 73	task_id, initial_assignee_ids = get_task_and_assignees(	×
74	planfix_client, contact_id
75	)
76	else:
NEW 77	self.send_json(	×
78	200,
79	{"answer": self.translate("task_and_assignee_together")},
80	)
81	return	×
82
NEW 83	access_report = grant_access(	×
84	planfix_client,
85	service,
86	drive_config,
87	role,
88	task_id,
89	initial_assignee_ids,
90	folder_id,
91	)
92	except LocalizedError as exc:	×
NEW 93	self.send_json(200, {"answer": self.translate(exc.key, **exc.context)})	×
UNCOV 94	return	×
95	except Exception as exc: # pylint: disable=broad-except	×
96	logger.exception("Failed to process request: %s", exc)	×
NEW 97	self.send_json(200, {"answer": self.translate("internal_server_error")})	×
UNCOV 98	return	×
99
100	granted_accounts = access_report["granted_accounts"]	×
101	existing_accounts = access_report["existing_accounts"]	×
NEW 102	answer = self.translate(	×
103	"granted_existing",
104	granted=self._format_accounts(granted_accounts),
105	existing=self._format_accounts(existing_accounts),
106	)
NEW 107	self.send_json(	×
108	200,
109	{
110	"answer": answer,
111	"granted_accounts": granted_accounts,
112	"existing_accounts": existing_accounts,
113	},
114	)
115
116	def _handle_create_client_folder(self, payload: Dict[str, Any]) -> None:	×
117	required_fields = ["contact_id", "folder_name"]	×
118	missing_fields = [	×
119	field for field in required_fields if field not in payload
120	]
121	if missing_fields:	×
NEW 122	self.send_json(	×
123	200,
124	{
125	"answer": self.translate(
126	"missing_fields", fields=", ".join(missing_fields)
127	)
128	},
129	)
130	return	×
131
132	try:	×
133	contact_id = int(payload["contact_id"])	×
134	folder_name = str(payload["folder_name"]).strip()	×
NEW 135	task_id, initial_assignee_ids = get_task_and_assignees(	×
136	planfix_client, contact_id
137	)
NEW 138	folder, created = create_client_folder(	×
139	service, drive_config, folder_name
140	)
NEW 141	if not created:	×
NEW 142	folder_url = (	×
143	f"https://drive.google.com/drive/folders/{folder['id']}"
144	)
NEW 145	self.send_json(	×
146	200,
147	{
148	"answer": self.translate(
149	"client_folder_exists", folder_url=folder_url
150	)
151	},
152	)
153	return	×
154
NEW 155	access_report = grant_access(	×
156	planfix_client,
157	service,
158	drive_config,
159	role,
160	task_id,
161	initial_assignee_ids,
162	folder["id"],
163	)
164	except LocalizedError as exc:	×
NEW 165	self.send_json(200, {"answer": self.translate(exc.key, **exc.context)})	×
UNCOV 166	return	×
167	except Exception as exc: # pylint: disable=broad-except	×
168	logger.exception("Failed to process request: %s", exc)	×
NEW 169	self.send_json(200, {"answer": self.translate("internal_server_error")})	×
UNCOV 170	return	×
171
172	granted_accounts = access_report["granted_accounts"]	×
173	existing_accounts = access_report["existing_accounts"]	×
NEW 174	answer = self.translate(	×
175	"granted_existing",
176	granted=self._format_accounts(granted_accounts),
177	existing=self._format_accounts(existing_accounts),
178	)
179	folder_url = f"https://drive.google.com/drive/folders/{folder['id']}"	×
NEW 180	self.send_json(	×
181	200,
182	{
183	"answer": self.translate(
184	"folder_created",
185	folder_name=folder_name,
186	details=answer,
187	folder_url=folder_url,
188	),
189	"folder_id": folder["id"],
190	"folder_url": folder_url,
191	"granted_accounts": granted_accounts,
192	"existing_accounts": existing_accounts,
193	},
194	)
195
196	def do_POST(self) -> None: # noqa: N802	×
197	if self.path == "/set_client_folder_access":	×
NEW 198	if not self.authenticate():	×
199	return	×
200
201	try:	×
NEW 202	payload = self.parse_json_body()	×
203	except LocalizedError as exc:	×
NEW 204	self.send_json(	×
205	200, {"answer": self.translate(exc.key, **exc.context)}
206	)
207	return	×
208
209	self._log_request(payload)	×
210	self._handle_set_client_folder_access(payload)	×
211	return	×
212
213	if self.path == "/create_client_folder":	×
NEW 214	if not self.authenticate():	×
215	return	×
216
217	try:	×
NEW 218	payload = self.parse_json_body()	×
219	except LocalizedError as exc:	×
NEW 220	self.send_json(	×
221	200, {"answer": self.translate(exc.key, **exc.context)}
222	)
223	return	×
224
225	self._log_request(payload)	×
226	self._handle_create_client_folder(payload)	×
227	return	×
228
NEW 229	self.send_json(200, {"answer": translate(language, "not_found")})	×
230
231	return AccessHandler	×

popstas / google-drive-access / 19721176345

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous