Implemented the next missing parity slice around session-scoped environment variables.
This commit is contained in:
@@ -707,6 +707,7 @@ Done:
|
|||||||
- [x] Basic git status snapshot
|
- [x] Basic git status snapshot
|
||||||
- [x] Basic shell/subprocess handling
|
- [x] Basic shell/subprocess handling
|
||||||
- [x] Bundled small portable utilities — `utils/array.ts`, `utils/set.ts`, `utils/objectGroupBy.ts`, `utils/xml.ts`, `utils/uuid.ts` ported in `src/small_utils.py`
|
- [x] Bundled small portable utilities — `utils/array.ts`, `utils/set.ts`, `utils/objectGroupBy.ts`, `utils/xml.ts`, `utils/uuid.ts` ported in `src/small_utils.py`
|
||||||
|
- [x] Session-scoped env-var registry (`utils/sessionEnvVars.ts`) ported in `src/session_env_vars.py`
|
||||||
|
|
||||||
Missing major utility categories:
|
Missing major utility categories:
|
||||||
|
|
||||||
|
|||||||
@@ -0,0 +1,46 @@
|
|||||||
|
"""Session-scoped environment variables — Python port of
|
||||||
|
``utils/sessionEnvVars.ts``.
|
||||||
|
|
||||||
|
These are env vars set during a session (via the upstream ``/env`` slash
|
||||||
|
command in npm) and applied only to spawned child processes — not to the
|
||||||
|
host Python REPL/agent process itself. Bash and similar tool providers
|
||||||
|
read this map to merge into ``subprocess`` environments.
|
||||||
|
|
||||||
|
Mirrors the upstream module-level singleton: callers import the helpers
|
||||||
|
directly rather than passing a registry around.
|
||||||
|
"""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
from collections.abc import Mapping
|
||||||
|
from types import MappingProxyType
|
||||||
|
|
||||||
|
_session_env_vars: dict[str, str] = {}
|
||||||
|
|
||||||
|
|
||||||
|
def get_session_env_vars() -> Mapping[str, str]:
|
||||||
|
"""Return a read-only view of the current session env vars."""
|
||||||
|
return MappingProxyType(_session_env_vars)
|
||||||
|
|
||||||
|
|
||||||
|
def set_session_env_var(name: str, value: str) -> None:
|
||||||
|
"""Set ``name=value`` for the rest of this session's child processes."""
|
||||||
|
_session_env_vars[name] = value
|
||||||
|
|
||||||
|
|
||||||
|
def delete_session_env_var(name: str) -> None:
|
||||||
|
"""Remove ``name`` from the session env (no-op if absent)."""
|
||||||
|
_session_env_vars.pop(name, None)
|
||||||
|
|
||||||
|
|
||||||
|
def clear_session_env_vars() -> None:
|
||||||
|
"""Drop every session-scoped env var."""
|
||||||
|
_session_env_vars.clear()
|
||||||
|
|
||||||
|
|
||||||
|
__all__ = [
|
||||||
|
'get_session_env_vars',
|
||||||
|
'set_session_env_var',
|
||||||
|
'delete_session_env_var',
|
||||||
|
'clear_session_env_vars',
|
||||||
|
]
|
||||||
@@ -0,0 +1,63 @@
|
|||||||
|
"""Tests for ``src/session_env_vars.py``."""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import unittest
|
||||||
|
|
||||||
|
from src.session_env_vars import (
|
||||||
|
clear_session_env_vars,
|
||||||
|
delete_session_env_var,
|
||||||
|
get_session_env_vars,
|
||||||
|
set_session_env_var,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
class SessionEnvVarsTest(unittest.TestCase):
|
||||||
|
def setUp(self) -> None:
|
||||||
|
clear_session_env_vars()
|
||||||
|
|
||||||
|
def tearDown(self) -> None:
|
||||||
|
clear_session_env_vars()
|
||||||
|
|
||||||
|
def test_starts_empty(self) -> None:
|
||||||
|
self.assertEqual(dict(get_session_env_vars()), {})
|
||||||
|
|
||||||
|
def test_set_and_get(self) -> None:
|
||||||
|
set_session_env_var('FOO', 'bar')
|
||||||
|
self.assertEqual(get_session_env_vars()['FOO'], 'bar')
|
||||||
|
|
||||||
|
def test_set_overwrites_existing(self) -> None:
|
||||||
|
set_session_env_var('FOO', 'one')
|
||||||
|
set_session_env_var('FOO', 'two')
|
||||||
|
self.assertEqual(get_session_env_vars()['FOO'], 'two')
|
||||||
|
|
||||||
|
def test_delete_removes(self) -> None:
|
||||||
|
set_session_env_var('FOO', 'bar')
|
||||||
|
delete_session_env_var('FOO')
|
||||||
|
self.assertNotIn('FOO', get_session_env_vars())
|
||||||
|
|
||||||
|
def test_delete_missing_is_noop(self) -> None:
|
||||||
|
delete_session_env_var('NEVER_SET')
|
||||||
|
self.assertEqual(dict(get_session_env_vars()), {})
|
||||||
|
|
||||||
|
def test_clear_drops_everything(self) -> None:
|
||||||
|
set_session_env_var('A', '1')
|
||||||
|
set_session_env_var('B', '2')
|
||||||
|
clear_session_env_vars()
|
||||||
|
self.assertEqual(dict(get_session_env_vars()), {})
|
||||||
|
|
||||||
|
def test_returned_mapping_is_read_only(self) -> None:
|
||||||
|
set_session_env_var('FOO', 'bar')
|
||||||
|
view = get_session_env_vars()
|
||||||
|
with self.assertRaises(TypeError):
|
||||||
|
view['FOO'] = 'mutated' # type: ignore[index]
|
||||||
|
|
||||||
|
def test_view_reflects_subsequent_mutations(self) -> None:
|
||||||
|
view = get_session_env_vars()
|
||||||
|
self.assertNotIn('FOO', view)
|
||||||
|
set_session_env_var('FOO', 'bar')
|
||||||
|
self.assertEqual(view['FOO'], 'bar')
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == '__main__':
|
||||||
|
unittest.main()
|
||||||
Reference in New Issue
Block a user