diff --git a/src/volttrontesting/platformwrapper.py b/src/volttrontesting/platformwrapper.py index 18d2890..1d6a40b 100644 --- a/src/volttrontesting/platformwrapper.py +++ b/src/volttrontesting/platformwrapper.py @@ -492,6 +492,31 @@ def build_agent(self, identity: Identity, agent_class: AgentT = Agent, options: return agent + def run_command(self, cmd: list, cwd: Path | str = None) -> str: + """ + Execute a shell command within the virtual environment. This will run + in the platformwrapper's context. + + if cwd is not set then the cwd will be set to `self.volttron_home` + + :raises CalledProcessError: If subprocess return value is not 0. + :param cmd: list passed to subprocess + :param cwd: directory to run the command in. + :return: response of the call. + """ + if cwd is None: + cwd = self.volttron_home + elif isinstance(cwd, Path): + cwd = cwd.as_posix() + + try: + output = self._virtual_env.run(args=cmd, capture=True, cwd=cwd, env=self._platform_environment, text=True) + except CalledProcessError as e: + print(f"Error:\n{e.output}") + raise + + return output + def install_library(self, library: str | Path, version: str = "latest"): if isinstance(library, Path): @@ -504,7 +529,8 @@ def install_library(self, library: str | Path, version: str = "latest"): cmd = f"poetry add {library}@latest" try: - output = self._virtual_env.run(args=cmd, capture=True, cwd=self.volttron_home) + output = self._virtual_env.run(args=cmd, env=self._platform_environment, capture=True, + cwd=self.volttron_home) except CalledProcessError as e: print(f"Error:\n{e.output}") raise diff --git a/tests/test_platformwrapper.py b/tests/test_platformwrapper.py index 7290363..525ab1f 100644 --- a/tests/test_platformwrapper.py +++ b/tests/test_platformwrapper.py @@ -31,6 +31,25 @@ from pathlib import Path +def test_run_command(get_pyproject_toml): + p = None + try: + options = create_server_options() + + p = PlatformWrapper(options=options, project_toml_file=get_pyproject_toml) + cmd = 'ls -la'.split() + output = p.run_command(cmd=cmd).split("\n") + + try: + next(filter(lambda a: a.find("config") > 0, output)) + except StopIteration: + pytest.fail("Couldn't find config in directory!") + + finally: + p.shutdown_platform() + p.cleanup() + + def test_can_enable_sys_queue(get_pyproject_toml): p = None try: