21945463160

Committed 12 Feb 2026 11:51AM UTC coverage: 59.669% (+0.2%) from 59.491%

Build # 21945463160

Build Type

push

github

Committed by

popstas

Commit Message

fix(http): normalize lang list and case

Unwrap list-valued lang field (["Ru"] → "ru"), lowercase it, and
fall back to config default for unsupported values.

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

Run Details

9 of 13 new or added lines in 1 file covered. (69.23%)

1623 of 2720 relevant lines covered (59.67%)

0.6 hits per line

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

42.65

/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 _apply_lang(self, payload: Dict[str, Any]) -> None:
            if "lang" not in payload:
                return
            lang = payload["lang"]
            if isinstance(lang, list):
                lang = lang[0] if lang else None
            if isinstance(lang, str):
                lang = lang.lower()
            if lang in ("ru", "en"):
                self.language = lang

        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

                self._apply_lang(payload)
                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

                self._apply_lang(payload)
                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

                self._apply_lang(payload)
                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 _apply_lang(self, payload: Dict[str, Any]) -> None:	1✔
37	if "lang" not in payload:	1✔
NEW 38	return	×
39	lang = payload["lang"]	1✔
40	if isinstance(lang, list):	1✔
41	lang = lang[0] if lang else None	1✔
42	if isinstance(lang, str):	1✔
43	lang = lang.lower()	1✔
44	if lang in ("ru", "en"):	1✔
45	self.language = lang	1✔
46
47	def _format_accounts(self, accounts: List[str]) -> str:	1✔
48	return ", ".join(accounts) if accounts else self.translate("none")	1✔
49
50	def do_POST(self) -> None: # noqa: N802	1✔
51	if self.path == "/set_client_folder_access":	×
52	if not self.authenticate():	×
53	return	×
54
55	try:	×
56	payload = self.parse_json_body()	×
57	except LocalizedError as exc:	×
58	self.send_json(	×
59	200, {"answer": self.translate(exc.key, **exc.context)}
60	)
61	return	×
62
NEW 63	self._apply_lang(payload)	×
64	self._log_request(payload)	×
65	set_client_folder_access_route.handle(	×
66	self,
67	payload,
68	planfix_client=planfix_client,
69	service=service,
70	drive_config=drive_config,
71	role=role,
72	)
73	return	×
74
75	if self.path == "/create_client_folder":	×
76	if not self.authenticate():	×
77	return	×
78
79	try:	×
80	payload = self.parse_json_body()	×
81	except LocalizedError as exc:	×
82	self.send_json(	×
83	200, {"answer": self.translate(exc.key, **exc.context)}
84	)
85	return	×
86
NEW 87	self._apply_lang(payload)	×
88	self._log_request(payload)	×
89	create_client_folder_route.handle(	×
90	self,
91	payload,
92	planfix_client=planfix_client,
93	service=service,
94	drive_config=drive_config,
95	role=role,
96	)
97	return	×
98
99	if self.path == "/share_file":	×
100	if not self.authenticate():	×
101	return	×
102
103	try:	×
104	payload = self.parse_json_body()	×
105	except LocalizedError as exc:	×
106	self.send_json(	×
107	200, {"answer": self.translate(exc.key, **exc.context)}
108	)
109	return	×
110
NEW 111	self._apply_lang(payload)	×
112	self._log_request(payload)	×
113	share_file_route.handle(	×
114	self,
115	payload,
116	service=service,
117	drive_config=drive_config,
118	share_file_config=share_file_config,
119	)
120	return	×
121
122	self.send_json(200, {"answer": self.translate("not_found")})	×
123
124	return AccessHandler	1✔

popstas / google-drive-access / 21945463160

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