Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
72 changes: 56 additions & 16 deletions src/powercontext/cli/env_file.py
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand All @@ -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
Expand All @@ -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]:
Expand All @@ -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]
Expand Down Expand Up @@ -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
Expand All @@ -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]:
Expand Down
22 changes: 22 additions & 0 deletions tests/test_config_cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -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(
Expand Down
43 changes: 43 additions & 0 deletions tests/test_env_file.py
Original file line number Diff line number Diff line change
Expand Up @@ -14,7 +14,9 @@

from __future__ import annotations

import json
import os
import time

import pytest

Expand All @@ -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"}

Expand Down
Loading