21943951424

Committed 12 Feb 2026 10:58AM UTC coverage: 59.491% (-0.04%) from 59.534%

Build # 21943951424

Build Type

push

github

Committed by

popstas

Commit Message

feat(http): add folder_url to response, lang arg

Add folder_url to granted_existing translation and JSON response
in set_client_folder_access and create_client_folder routes.

Add per-request lang override via payload field across all routes.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Run Details

2 of 8 new or added lines in 3 files covered. (25.0%)

1 existing line in 1 file now uncovered.

1614 of 2713 relevant lines covered (59.49%)

0.59 hits per line

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

32.79

/src/drive_audit/http/handler.py

"""HTTP handler for managing Google Drive access."""

import json
from typing import Any, Dict, List

from loguru import logger

from ..http_utils import JsonRequestHandler, LocalizedError
from ..model import DriveConfig, HttpConfig
from ..planfix_client import PlanfixClient
from . import create_client_folder as create_client_folder_route
from . import set_client_folder_access as set_client_folder_access_route
from . import share_file as share_file_route


def create_handler(
    planfix_client: PlanfixClient,
    service,
    http_config: HttpConfig,
    drive_config: DriveConfig,
    role: str,
    share_file_config=None,
):
    handler_http_config = http_config
    handler_language = http_config.lang

    class AccessHandler(JsonRequestHandler):
        http_config = handler_http_config
        language = handler_language

        def _log_request(self, payload: Dict[str, Any]) -> None:
            logger.info(
                "{} request: {}", self.path, json.dumps(payload, ensure_ascii=False)
            )

        def _format_accounts(self, accounts: List[str]) -> str:
            return ", ".join(accounts) if accounts else self.translate("none")

        def do_POST(self) -> None:  # noqa: N802
            if self.path == "/set_client_folder_access":
                if not self.authenticate():
                    return

                try:
                    payload = self.parse_json_body()
                except LocalizedError as exc:
                    self.send_json(
                        200, {"answer": self.translate(exc.key, **exc.context)}
                    )
                    return

                if "lang" in payload:
                    self.language = payload["lang"]
                self._log_request(payload)
                set_client_folder_access_route.handle(
                    self,
                    payload,
                    planfix_client=planfix_client,
                    service=service,
                    drive_config=drive_config,
                    role=role,
                )
                return

            if self.path == "/create_client_folder":
                if not self.authenticate():
                    return

                try:
                    payload = self.parse_json_body()
                except LocalizedError as exc:
                    self.send_json(
                        200, {"answer": self.translate(exc.key, **exc.context)}
                    )
                    return

                if "lang" in payload:
                    self.language = payload["lang"]
                self._log_request(payload)
                create_client_folder_route.handle(
                    self,
                    payload,
                    planfix_client=planfix_client,
                    service=service,
                    drive_config=drive_config,
                    role=role,
                )
                return

            if self.path == "/share_file":
                if not self.authenticate():
                    return

                try:
                    payload = self.parse_json_body()
                except LocalizedError as exc:
                    self.send_json(
                        200, {"answer": self.translate(exc.key, **exc.context)}
                    )
                    return

                if "lang" in payload:
                    self.language = payload["lang"]
                self._log_request(payload)
                share_file_route.handle(
                    self,
                    payload,
                    service=service,
                    drive_config=drive_config,
                    share_file_config=share_file_config,
                )
                return

            self.send_json(200, {"answer": self.translate("not_found")})

    return AccessHandler

1	"""HTTP handler for managing Google Drive access."""
2
3	import json	1✔
4	from typing import Any, Dict, List	1✔
5
6	from loguru import logger	1✔
7
8	from ..http_utils import JsonRequestHandler, LocalizedError	1✔
9	from ..model import DriveConfig, HttpConfig	1✔
10	from ..planfix_client import PlanfixClient	1✔
11	from . import create_client_folder as create_client_folder_route	1✔
12	from . import set_client_folder_access as set_client_folder_access_route	1✔
13	from . import share_file as share_file_route	1✔
14
15
16	def create_handler(	1✔
17	planfix_client: PlanfixClient,
18	service,
19	http_config: HttpConfig,
20	drive_config: DriveConfig,
21	role: str,
22	share_file_config=None,
23	):
24	handler_http_config = http_config	1✔
25	handler_language = http_config.lang	1✔
26
27	class AccessHandler(JsonRequestHandler):	1✔
28	http_config = handler_http_config	1✔
29	language = handler_language	1✔
30
31	def _log_request(self, payload: Dict[str, Any]) -> None:	1✔
32	logger.info(	×
33	"{} request: {}", self.path, json.dumps(payload, ensure_ascii=False)
34	)
35
36	def _format_accounts(self, accounts: List[str]) -> str:	1✔
37	return ", ".join(accounts) if accounts else self.translate("none")	1✔
38
39	def do_POST(self) -> None: # noqa: N802	1✔
40	if self.path == "/set_client_folder_access":	×
41	if not self.authenticate():	×
42	return	×
43
44	try:	×
45	payload = self.parse_json_body()	×
46	except LocalizedError as exc:	×
47	self.send_json(	×
48	200, {"answer": self.translate(exc.key, **exc.context)}
49	)
50	return	×
51
NEW 52	if "lang" in payload:	×
NEW 53	self.language = payload["lang"]	×
54	self._log_request(payload)	×
55	set_client_folder_access_route.handle(	×
56	self,
57	payload,
58	planfix_client=planfix_client,
59	service=service,
60	drive_config=drive_config,
61	role=role,
62	)
63	return	×
64
65	if self.path == "/create_client_folder":	×
66	if not self.authenticate():	×
67	return	×
68
69	try:	×
70	payload = self.parse_json_body()	×
71	except LocalizedError as exc:	×
72	self.send_json(	×
73	200, {"answer": self.translate(exc.key, **exc.context)}
74	)
75	return	×
76
NEW 77	if "lang" in payload:	×
NEW 78	self.language = payload["lang"]	×
79	self._log_request(payload)	×
80	create_client_folder_route.handle(	×
81	self,
82	payload,
83	planfix_client=planfix_client,
84	service=service,
85	drive_config=drive_config,
86	role=role,
87	)
88	return	×
89
90	if self.path == "/share_file":	×
91	if not self.authenticate():	×
92	return	×
93
94	try:	×
95	payload = self.parse_json_body()	×
96	except LocalizedError as exc:	×
97	self.send_json(	×
98	200, {"answer": self.translate(exc.key, **exc.context)}
99	)
100	return	×
101
NEW 102	if "lang" in payload:	×
NEW 103	self.language = payload["lang"]	×
104	self._log_request(payload)	×
105	share_file_route.handle(	×
106	self,
107	payload,
108	service=service,
109	drive_config=drive_config,
110	share_file_config=share_file_config,
111	)
112	return	×
113
114	self.send_json(200, {"answer": self.translate("not_found")})	×
115
116	return AccessHandler	1✔

popstas / google-drive-access / 21943951424

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