Skip to content
Snippets Groups Projects
Commit 64d41486 authored by dilawar's avatar dilawar :ant:
Browse files

feat: fixes to the tests.

parent e622cad0
No related branches found
No related tags found
1 merge request!6Hotfix: fix to assert error post installation
Pipeline #3783 failed with stages
in 1 minute and 39 seconds
......@@ -54,7 +54,10 @@ def session(func):
@app.command("create-container")
@session
def create_remote_container(
user_input, *, recreate: bool = False, output_lines: T.List[str] = []
user_input,
*,
recreate: bool = False,
output_lines: T.List[str] = [],
):
"""Create container for the pipeline. The container starts running
immediately on the server. Use command `logs` to stream the output.
......@@ -71,9 +74,12 @@ def create_remote_container(
stream=True,
)
res.raise_for_status()
for line in track(
res.iter_lines(), description="BiTIA is setting up required infra..."
):
_lines = (
track(res.iter_lines(), description="BiTIA is setting up required infra...")
if not bconfig.get_config("plain", default=False)
else res.iter_lines()
)
for line in _lines:
output_lines.append(line.decode().rstrip())
logger.info(output_lines[-1])
return res
......@@ -154,11 +160,14 @@ def main(
verbose: VerbosityLevel = typer.Option(
VerbosityLevel.warning, case_sensitive=False
),
plain: bool = False,
server: T.Optional[str] = None,
):
"""
Callback
"""
bconfig.set_config("plain", plain)
bconfig.set_config("verbosity", verbose.value)
set_logger_level(verbose.value)
if server is not None:
......
......@@ -5,7 +5,17 @@ import tempfile
BITIA_MAIN_SCRIPT_NAME: T.Final[str] = "__main__.bitia.sh"
g_server = "https://public.bitia.link/api/v1"
_config = {"server": "https://public.bitia.link/api/v1"}
def set_config(key, val):
global _config
_config[key] = val
def get_config(key, *, strict: bool = True, default=None):
return _config[key] if strict else _config.get(key, default)
def bitia_dir() -> Path:
......@@ -19,10 +29,9 @@ def get_server(use_env: bool = True) -> str:
"""Server to use"""
if use_env and os.environ.get("BITIA_SERVER") is not None:
return os.environ["BITIA_SERVER"]
return g_server
return get_config("server")
def set_server(server: str):
"""set bitia server"""
global g_server
g_server = server
set_config("server", server)
from difflib import SequenceMatcher
import bitia
import bitia.config as bconfig
import bitia.__main__
def test_sanity():
assert bitia.version()
def test_sanity(capsys):
version = bitia.version()
assert len(version) >= 3, version
def test_run_simple():
for line in bitia.__main__.run_user_input("ls -ltr /"):
if line:
print(111, line.decode().rstrip())
for line in bitia.__main__.run_user_input("ls -ltr /", rerun=True):
if line:
print(222, line.decode().rstrip())
if __name__ == "__main__":
test_sanity()
test_run_simple()
def test_run_simple(capsys):
# set the plain-text mode.
bconfig.set_config("plain", True)
bitia.__main__.run_user_input("ls -ltr /")
captured = capsys.readouterr() # reset the internal buffer.
l1 = captured.out
bitia.__main__.run_user_input("ls -ltr /", rerun=True)
captured = capsys.readouterr()
l2 = captured.out
assert SequenceMatcher(a=l1, b=l2).ratio() > 0.plain9
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment