oxbrbtx/PrimAITE
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

#2533: set default dev output path + clean up

Browse Source
This commit is contained in:
Czar Echavez
2024-05-01 14:02:40 +01:00
parent 95643d3255
commit a2fb04e6f6
9 changed files with 84 additions and 95 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
.gitignore vendored
View File

@@ -151,6 +151,7 @@ docs/source/primaite-dependencies.rst
# outputs
src/primaite/outputs/
simulation_output/
sessions/
# benchmark session outputs
benchmark/output

11
src/primaite/__init__.py
View File

@@ -122,20 +122,13 @@ class _PrimaitePaths:
PRIMAITE_PATHS: Final[_PrimaitePaths] = _PrimaitePaths()
def _host_primaite_config() -> None:
if not PRIMAITE_PATHS.app_config_file_path.exists():
pkg_config_path = Path(pkg_resources.resource_filename("primaite", "setup/_package_data/primaite_config.yaml"))
shutil.copy2(pkg_config_path, PRIMAITE_PATHS.app_config_file_path)
_host_primaite_config()
def _get_primaite_config() -> Dict:
config_path = PRIMAITE_PATHS.app_config_file_path
if not config_path.exists():
# load from package if config does not exist
config_path = Path(pkg_resources.resource_filename("primaite", "setup/_package_data/primaite_config.yaml"))
# generate app config
shutil.copy2(config_path, PRIMAITE_PATHS.app_config_file_path)
with open(config_path, "r") as file:
# load from config
primaite_config = yaml.safe_load(file)

17
src/primaite/session/io.py
View File

@@ -5,8 +5,9 @@ from typing import Dict, List, Optional
from pydantic import BaseModel, ConfigDict
from primaite import getLogger, PRIMAITE_PATHS
from primaite import _PRIMAITE_ROOT, getLogger, PRIMAITE_PATHS
from primaite.simulator import LogLevel, SIM_OUTPUT
from primaite.utils.cli.primaite_config_utils import is_dev_mode
_LOGGER = getLogger(__name__)
@@ -61,17 +62,13 @@ class PrimaiteIO:
date_str = timestamp.strftime("%Y-%m-%d")
time_str = timestamp.strftime("%H-%M-%S")
# check if running in dev mode
# if is_dev_mode():
# # if dev mode, simulation output will be the repository root or whichever path is configured
# app_config = get_primaite_config_dict()
# if app_config["developer_mode"]["output_dir"] is not None:
# session_path = app_config["developer_mode"]["output_dir"]
# else:
# session_path = _PRIMAITE_ROOT.parent.parent / "simulation_output" / date_str / time_str
# else:
session_path = PRIMAITE_PATHS.user_sessions_path / date_str / time_str
# check if running in dev mode
if is_dev_mode():
# check if there is an output directory set in config
session_path = _PRIMAITE_ROOT.parent.parent / "sessions" / date_str / time_str
session_path.mkdir(exist_ok=True, parents=True)
return session_path

16
src/primaite/simulator/__init__.py
View File

@@ -3,7 +3,7 @@ from datetime import datetime
from enum import IntEnum
from pathlib import Path
from primaite import _PRIMAITE_ROOT, PRIMAITE_CONFIG
from primaite import _PRIMAITE_ROOT, PRIMAITE_CONFIG, PRIMAITE_PATHS
__all__ = ["SIM_OUTPUT"]
@@ -27,9 +27,17 @@ class LogLevel(IntEnum):
class _SimOutput:
def __init__(self):
self._path: Path = (
_PRIMAITE_ROOT.parent.parent / "simulation_output" / datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
)
date_str = datetime.now().strftime("%Y-%m-%d")
time_str = datetime.now().strftime("%H-%M-%S")
path = PRIMAITE_PATHS.user_sessions_path / date_str / time_str
if is_dev_mode():
# if dev mode is enabled, if output dir is not set, print to primaite repo root
path: Path = _PRIMAITE_ROOT.parent.parent / "sessions" / date_str / time_str / "simulation_output"
# otherwise print to output dir
self._path = path
self._save_pcap_logs: bool = False
self._save_sys_logs: bool = False
self._write_sys_log_to_terminal: bool = False

6
src/primaite/utils/cli/dev_cli.py
View File

@@ -15,7 +15,7 @@ PRODUCTION_MODE_MESSAGE = (
" :monkey_face::monkey_face::monkey_face: [/green]\n"
)
DEVELOPMENT_MODE_MESSAGE = (
DEVELOPER_MODE_MESSAGE = (
"\n[yellow] :construction::construction::construction: "
" PrimAITE is running in Development mode "
" :construction::construction::construction: [/yellow]\n"
@@ -38,7 +38,7 @@ def dev_mode():
def show():
"""Show if PrimAITE is in development mode or production mode."""
# print if dev mode is enabled
print(DEVELOPMENT_MODE_MESSAGE if is_dev_mode() else PRODUCTION_MODE_MESSAGE)
print(DEVELOPER_MODE_MESSAGE if is_dev_mode() else PRODUCTION_MODE_MESSAGE)
print("\nTo see available options, use [medium_turquoise]`primaite dev-mode --help`[/medium_turquoise]\n")
@@ -48,7 +48,7 @@ def enable():
# enable dev mode
PRIMAITE_CONFIG["developer_mode"]["enabled"] = True
update_primaite_application_config()
print(DEVELOPMENT_MODE_MESSAGE)
print(DEVELOPER_MODE_MESSAGE)
@dev.command()

6
src/primaite/utils/cli/primaite_config_utils.py
View File

@@ -5,11 +5,7 @@ from primaite import PRIMAITE_CONFIG, PRIMAITE_PATHS
def is_dev_mode() -> bool:
"""Returns True if PrimAITE is currently running in developer mode."""
return (
PRIMAITE_CONFIG["developer_mode"]["enabled"]
if (PRIMAITE_CONFIG.get("developer_mode", {}).get("enabled"))
else False
)
return PRIMAITE_CONFIG["developer_mode"]["enabled"]
def update_primaite_application_config() -> None:

15
tests/conftest.py
View File

@@ -40,21 +40,6 @@ ACTION_SPACE_NODE_ACTION_VALUES = 1
_LOGGER = getLogger(__name__)
@pytest.fixture(scope="function", autouse=True)
def set_syslog_output_to_true():
"""Will be run before each test."""
monkeypatch = MonkeyPatch()
monkeypatch.setattr(
SIM_OUTPUT,
"path",
Path(TEST_ASSETS_ROOT.parent.parent / "simulation_output" / datetime.now().strftime("%Y-%m-%d_%H-%M-%S")),
)
monkeypatch.setattr(SIM_OUTPUT, "save_pcap_logs", False)
monkeypatch.setattr(SIM_OUTPUT, "save_sys_logs", False)
yield
class TestService(Service):
"""Test Service class"""

97
tests/integration_tests/cli/test_dev_cli.py
View File

@@ -5,34 +5,47 @@ from pathlib import Path
import pkg_resources
import pytest
import yaml
from _pytest.monkeypatch import MonkeyPatch
from primaite import _PRIMAITE_ROOT, PRIMAITE_PATHS
from primaite.utils.cli.primaite_config_utils import get_primaite_config_dict
import primaite
from tests.integration_tests.cli import cli
@pytest.fixture(autouse=True)
def test_setup():
"""
Setup this test by copying the
Setup this test by using the default primaite app config in package
"""
original_config_path = PRIMAITE_PATHS.app_config_file_path # keep track of app config before test
current_config = primaite.PRIMAITE_CONFIG # store the config before test
original_config_path = primaite.PRIMAITE_PATHS.app_config_file_path # keep track of app config before test
temp_dir = tempfile.gettempdir()
temp_config = Path(temp_dir) / "primaite_config.yaml"
pkg_config_path = Path(pkg_resources.resource_filename("primaite", "setup/_package_data/primaite_config.yaml"))
shutil.copyfile(pkg_config_path, temp_config) # copy the default primaite config to temp directory
PRIMAITE_PATHS.app_config_file_path = temp_config # use the copy for the test
yield # run test
primaite.PRIMAITE_PATHS.app_config_file_path = temp_config # use the copy for the test
with open(pkg_config_path, "r") as file:
# load from config
config_dict = yaml.safe_load(file)
primaite.PRIMAITE_CONFIG = config_dict
assert primaite.PRIMAITE_CONFIG == config_dict
yield
os.remove(temp_config) # clean up temp file
PRIMAITE_PATHS.app_config_file_path = original_config_path # restore app conf because other devs will yell at me
primaite.PRIMAITE_CONFIG = current_config # restore config to prevent being yelled at
assert primaite.PRIMAITE_CONFIG == current_config
primaite.PRIMAITE_PATHS.app_config_file_path = original_config_path
@pytest.mark.skip(reason="borked")
def test_dev_mode_enable_disable():
"""Test dev mode enable and disable."""
# check defaults
config_dict = get_primaite_config_dict()
assert config_dict["developer_mode"]["enabled"] is False # not enabled by default
assert primaite.PRIMAITE_CONFIG["developer_mode"]["enabled"] is False # not enabled by default
result = cli(["dev-mode", "show"])
assert "Production" in result.output # should print that it is in Production mode by default
@@ -41,8 +54,7 @@ def test_dev_mode_enable_disable():
assert "Development" in result.output # should print that it is in Development mode
config_dict = get_primaite_config_dict()
assert config_dict["developer_mode"]["enabled"] # config should reflect that dev mode is enabled
assert primaite.PRIMAITE_CONFIG["developer_mode"]["enabled"] # config should reflect that dev mode is enabled
result = cli(["dev-mode", "show"])
assert "Development" in result.output # should print that it is in Development mode
@@ -51,136 +63,123 @@ def test_dev_mode_enable_disable():
assert "Production" in result.output # should print that it is in Production mode
config_dict = get_primaite_config_dict()
assert config_dict["developer_mode"]["enabled"] is False # config should reflect that dev mode is disabled
assert (
primaite.PRIMAITE_CONFIG["developer_mode"]["enabled"] is False
) # config should reflect that dev mode is disabled
result = cli(["dev-mode", "show"])
assert "Production" in result.output # should print that it is in Production mode
@pytest.mark.skip(reason="borked")
def test_dev_mode_config_sys_log_level():
"""Check that the system log level can be changed via CLI."""
# check defaults
config_dict = get_primaite_config_dict()
assert config_dict["developer_mode"]["sys_log_level"] == "DEBUG" # DEBUG by default
assert primaite.PRIMAITE_CONFIG["developer_mode"]["sys_log_level"] == "DEBUG" # DEBUG by default
result = cli(["dev-mode", "config", "-level", "WARNING"])
assert "sys_log_level=WARNING" in result.output # should print correct value
config_dict = get_primaite_config_dict()
# config should reflect that log level is WARNING
assert config_dict["developer_mode"]["sys_log_level"] == "WARNING"
assert primaite.PRIMAITE_CONFIG["developer_mode"]["sys_log_level"] == "WARNING"
result = cli(["dev-mode", "config", "--sys-log-level", "INFO"])
assert "sys_log_level=INFO" in result.output # should print correct value
config_dict = get_primaite_config_dict()
# config should reflect that log level is WARNING
assert config_dict["developer_mode"]["sys_log_level"] == "INFO"
assert primaite.PRIMAITE_CONFIG["developer_mode"]["sys_log_level"] == "INFO"
@pytest.mark.skip(reason="borked")
def test_dev_mode_config_sys_logs_enable_disable():
"""Test that the system logs output can be enabled or disabled."""
# check defaults
config_dict = get_primaite_config_dict()
assert config_dict["developer_mode"]["output_sys_logs"] is False # False by default
assert primaite.PRIMAITE_CONFIG["developer_mode"]["output_sys_logs"] is False # False by default
result = cli(["dev-mode", "config", "--output-sys-logs"])
assert "output_sys_logs=True" in result.output # should print correct value
config_dict = get_primaite_config_dict()
# config should reflect that output_sys_logs is True
assert config_dict["developer_mode"]["output_sys_logs"]
assert primaite.PRIMAITE_CONFIG["developer_mode"]["output_sys_logs"]
result = cli(["dev-mode", "config", "--no-sys-logs"])
assert "output_sys_logs=False" in result.output # should print correct value
config_dict = get_primaite_config_dict()
# config should reflect that output_sys_logs is True
assert config_dict["developer_mode"]["output_sys_logs"] is False
assert primaite.PRIMAITE_CONFIG["developer_mode"]["output_sys_logs"] is False
result = cli(["dev-mode", "config", "-sys"])
assert "output_sys_logs=True" in result.output # should print correct value
config_dict = get_primaite_config_dict()
# config should reflect that output_sys_logs is True
assert config_dict["developer_mode"]["output_sys_logs"]
assert primaite.PRIMAITE_CONFIG["developer_mode"]["output_sys_logs"]
result = cli(["dev-mode", "config", "-nsys"])
assert "output_sys_logs=False" in result.output # should print correct value
config_dict = get_primaite_config_dict()
# config should reflect that output_sys_logs is True
assert config_dict["developer_mode"]["output_sys_logs"] is False
assert primaite.PRIMAITE_CONFIG["developer_mode"]["output_sys_logs"] is False
@pytest.mark.skip(reason="borked")
def test_dev_mode_config_pcap_logs_enable_disable():
"""Test that the pcap logs output can be enabled or disabled."""
# check defaults
config_dict = get_primaite_config_dict()
assert config_dict["developer_mode"]["output_pcap_logs"] is False # False by default
assert primaite.PRIMAITE_CONFIG["developer_mode"]["output_pcap_logs"] is False # False by default
result = cli(["dev-mode", "config", "--output-pcap-logs"])
assert "output_pcap_logs=True" in result.output # should print correct value
config_dict = get_primaite_config_dict()
# config should reflect that output_pcap_logs is True
assert config_dict["developer_mode"]["output_pcap_logs"]
assert primaite.PRIMAITE_CONFIG["developer_mode"]["output_pcap_logs"]
result = cli(["dev-mode", "config", "--no-pcap-logs"])
assert "output_pcap_logs=False" in result.output # should print correct value
config_dict = get_primaite_config_dict()
# config should reflect that output_pcap_logs is True
assert config_dict["developer_mode"]["output_pcap_logs"] is False
assert primaite.PRIMAITE_CONFIG["developer_mode"]["output_pcap_logs"] is False
result = cli(["dev-mode", "config", "-pcap"])
assert "output_pcap_logs=True" in result.output # should print correct value
config_dict = get_primaite_config_dict()
# config should reflect that output_pcap_logs is True
assert config_dict["developer_mode"]["output_pcap_logs"]
assert primaite.PRIMAITE_CONFIG["developer_mode"]["output_pcap_logs"]
result = cli(["dev-mode", "config", "-npcap"])
assert "output_pcap_logs=False" in result.output # should print correct value
config_dict = get_primaite_config_dict()
# config should reflect that output_pcap_logs is True
assert config_dict["developer_mode"]["output_pcap_logs"] is False
assert primaite.PRIMAITE_CONFIG["developer_mode"]["output_pcap_logs"] is False
@pytest.mark.skip(reason="borked")
def test_dev_mode_config_output_to_terminal_enable_disable():
"""Test that the output to terminal can be enabled or disabled."""
# check defaults
config_dict = get_primaite_config_dict()
assert config_dict["developer_mode"]["output_to_terminal"] is False # False by default
assert primaite.PRIMAITE_CONFIG["developer_mode"]["output_to_terminal"] is False # False by default
result = cli(["dev-mode", "config", "--output-to-terminal"])
assert "output_to_terminal=True" in result.output # should print correct value
config_dict = get_primaite_config_dict()
# config should reflect that output_to_terminal is True
assert config_dict["developer_mode"]["output_to_terminal"]
assert primaite.PRIMAITE_CONFIG["developer_mode"]["output_to_terminal"]
result = cli(["dev-mode", "config", "--no-terminal"])
assert "output_to_terminal=False" in result.output # should print correct value
config_dict = get_primaite_config_dict()
# config should reflect that output_to_terminal is True
assert config_dict["developer_mode"]["output_to_terminal"] is False
assert primaite.PRIMAITE_CONFIG["developer_mode"]["output_to_terminal"] is False
result = cli(["dev-mode", "config", "-t"])
assert "output_to_terminal=True" in result.output # should print correct value
config_dict = get_primaite_config_dict()
# config should reflect that output_to_terminal is True
assert config_dict["developer_mode"]["output_to_terminal"]
assert primaite.PRIMAITE_CONFIG["developer_mode"]["output_to_terminal"]
result = cli(["dev-mode", "config", "-nt"])
assert "output_to_terminal=False" in result.output # should print correct value
config_dict = get_primaite_config_dict()
# config should reflect that output_to_terminal is True
assert config_dict["developer_mode"]["output_to_terminal"] is False
assert primaite.PRIMAITE_CONFIG["developer_mode"]["output_to_terminal"] is False

10
tests/unit_tests/_primaite/_simulator/_system/core/test_sys_log.py
View File

@@ -2,10 +2,20 @@ from uuid import uuid4
import pytest
from primaite import PRIMAITE_CONFIG
from primaite.simulator import LogLevel, SIM_OUTPUT
from primaite.simulator.system.core.sys_log import SysLog
@pytest.fixture(autouse=True)
def override_dev_mode_temporarily():
"""Temporarily turn off dev mode for this test."""
primaite_dev_mode = PRIMAITE_CONFIG["developer_mode"]["enabled"]
PRIMAITE_CONFIG["developer_mode"]["enabled"] = False
yield # run tests
PRIMAITE_CONFIG["developer_mode"]["enabled"] = primaite_dev_mode
@pytest.fixture(scope="function")
def syslog() -> SysLog:
return SysLog(hostname="test")