diff --git a/src/powercontext/cli/env_file.py b/src/powercontext/cli/env_file.py index 9ebf7453c..a65d4f927 100644 --- a/src/powercontext/cli/env_file.py +++ b/src/powercontext/cli/env_file.py @@ -18,7 +18,7 @@ import os import re -from collections.abc import Collection, Generator, Mapping, MutableMapping +from collections.abc import Collection, Generator, Iterator, Mapping, MutableMapping from contextlib import contextmanager from pathlib import Path @@ -44,7 +44,8 @@ def parse_environment(content: str, *, source: str = "environment") -> dict[str, """ environment: dict[str, str] = {} - for line_number, line in enumerate(content.splitlines(), start=1): + lines = iter(enumerate(content.splitlines(), start=1)) + for line_number, line in lines: stripped = line.lstrip(" \t") if not stripped or stripped.startswith("#"): continue @@ -53,12 +54,7 @@ def parse_environment(content: str, *, source: str = "environment") -> dict[str, stripped = stripped[export.end() :] if "\x00" in stripped: raise EnvironmentFileError(f"invalid NUL character at {source}:{line_number}") # noqa: TRY003 - try: - tokens = _split_shell_words(stripped) - except ValueError as error: - raise EnvironmentFileError( # noqa: TRY003 - f"invalid assignment at {source}:{line_number}: {error}" - ) from error + tokens = _split_assignment(stripped, lines, source=source, line_number=line_number) if not tokens: continue if len(tokens) != 1 or "=" not in tokens[0]: @@ -76,13 +72,61 @@ def parse_environment(content: str, *, source: str = "environment") -> dict[str, return environment -def _split_shell_words(line: str) -> list[str]: # noqa: C901 - """Split one shell assignment without expansion or command evaluation.""" +def _split_assignment( + first_line: str, + following_lines: Iterator[tuple[int, str]], + *, + source: str, + line_number: int, +) -> list[str]: + """Split one assignment, consuming physical lines until its quotes close.""" words: list[str] = [] word: list[str] = [] quote = "" word_started = False + line = first_line + while True: + try: + quote, word_started = _scan_shell_words( + line, + words, + word, + quote=quote, + word_started=word_started, + ) + except ValueError as error: + raise EnvironmentFileError( # noqa: TRY003 + f"invalid assignment at {source}:{line_number}: {error}" + ) from error + if not quote: + if word_started: + words.append("".join(word)) + return words + try: + continuation_line_number, line = next(following_lines) + except StopIteration: + error = ValueError(_NO_CLOSING_QUOTATION) + raise EnvironmentFileError( # noqa: TRY003 + f"invalid assignment at {source}:{line_number}: {error}" + ) from error + if "\x00" in line: + raise EnvironmentFileError( # noqa: TRY003 + f"invalid NUL character at {source}:{continuation_line_number}" + ) + word.append("\n") + + +def _scan_shell_words( # noqa: C901 + line: str, + words: list[str], + word: list[str], + *, + quote: str, + word_started: bool, +) -> tuple[str, bool]: + """Scan one physical line while carrying the current assignment state.""" + index = 0 while index < len(line): character = line[index] @@ -115,7 +159,7 @@ def _split_shell_words(line: str) -> list[str]: # noqa: C901 elif character in {" ", "\t"}: if word_started: words.append("".join(word)) - word = [] + word.clear() word_started = False elif character == "#" and not word_started: break @@ -125,11 +169,7 @@ def _split_shell_words(line: str) -> list[str]: # noqa: C901 word.append(character) word_started = True index += 1 - if quote: - raise ValueError(_NO_CLOSING_QUOTATION) - if word_started: - words.append("".join(word)) - return words + return quote, word_started def read_environment_file(path: Path) -> dict[str, str]: diff --git a/tests/test_config_cli.py b/tests/test_config_cli.py index 6d68f4382..0767f39f5 100644 --- a/tests/test_config_cli.py +++ b/tests/test_config_cli.py @@ -249,6 +249,28 @@ def test_validate_reports_invalid_numeric_values_without_a_traceback(tmp_path: P assert "Traceback" not in result.output +def test_validate_accepts_multiline_quoted_dashboard_scopes(tmp_path: Path) -> None: + environment = tmp_path / ".env" + multiline = """POWERCONTEXT_SERVER_DASHBOARD_SCOPES='[ + { + "scope_id": "project:quickstart", + "display_name": "Quick Start" + } +]'""" + generated = config_cli.render_managed_block(_configuration()) + content = "\n".join( + line for line in generated.splitlines() if not line.startswith("POWERCONTEXT_SERVER_DASHBOARD_SCOPES=") + ) + content = f"{content}\n{multiline}\n" + environment.write_text(content, encoding="utf-8") + + with patch.object(config_cli, "_validate_provider_models"): + result = CliRunner().invoke(config_cli.app, ["validate", "--env-file", str(environment)]) + + assert result.exit_code == 0 + assert "Configuration is valid" in result.output + + def test_show_redacts_standard_credential_container_variables(tmp_path: Path) -> None: environment = tmp_path / ".env" environment.write_text( diff --git a/tests/test_env_file.py b/tests/test_env_file.py index 7554a9678..34fa512e7 100644 --- a/tests/test_env_file.py +++ b/tests/test_env_file.py @@ -14,7 +14,9 @@ from __future__ import annotations +import json import os +import time import pytest @@ -39,6 +41,47 @@ def test_quoted_values_keep_hashes_and_spaces() -> None: assert parse_environment(content) == {"TOKEN": "#not a comment", "OTHER": "plain#tag"} +def test_multiline_quoted_json_value_is_preserved() -> None: + content = """POWERCONTEXT_SERVER_DASHBOARD_SCOPES='[ + { + "scope_id": "git:github.com/oceanbase/powercontext", + "display_name": "powercontext" + } +]' +OTHER=value +""" + + assert parse_environment(content) == { + "POWERCONTEXT_SERVER_DASHBOARD_SCOPES": """[ + { + "scope_id": "git:github.com/oceanbase/powercontext", + "display_name": "powercontext" + } +]""", + "OTHER": "value", + } + + +def test_multiline_double_quoted_value_unescapes_json_quotes() -> None: + content = 'JSON="[\n {\\"name\\": \\"value\\"}\n]"\n' + + assert parse_environment(content) == {"JSON": '[\n {"name": "value"}\n]'} + + +def test_large_multiline_value_parses_within_linear_time_budget() -> None: + value = json.dumps( + [{"scope_id": f"project:{index}", "display_name": f"Project {index}"} for index in range(800)], + indent=2, + ) + started = time.monotonic() + + parsed = parse_environment(f"SCOPES='{value}'\n") + + elapsed = time.monotonic() - started + assert parsed == {"SCOPES": value} + assert elapsed < 1.0, f"large multiline value took {elapsed:.3f}s to parse" + + def test_url_fragment_assignment_survives() -> None: assert parse_environment("URL=https://example.com/page#section\n") == {"URL": "https://example.com/page#section"}