From 50e76e8a3b4a2ecdb425e60fcbda893ad6de6fbb Mon Sep 17 00:00:00 2001 From: Enis Isufi Date: Wed, 20 May 2026 11:52:50 +0200 Subject: [PATCH 1/3] first draft adding new features Co-authored-by: Copilot --- tests/conftest.py | 11 ++++++--- tests/test_cmd_cubes.py | 2 +- tests/test_cmd_dimension.py | 2 +- tests/test_cmd_subset.py | 6 +++-- tests/test_cmd_view.py | 6 +++-- tm1cli/commands/cube.py | 16 ++++++++---- tm1cli/commands/dimension.py | 18 +++++++++----- tm1cli/commands/process.py | 21 +++++++++++++--- tm1cli/commands/subset.py | 40 +++++++++++++++++++++++++----- tm1cli/commands/view.py | 32 ++++++++++++++++++++---- tm1cli/utils/list_utils.py | 48 ++++++++++++++++++++++++++++++++++++ 11 files changed, 168 insertions(+), 34 deletions(-) create mode 100644 tm1cli/utils/list_utils.py diff --git a/tests/conftest.py b/tests/conftest.py index a347c38..a4d6af5 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -4,7 +4,7 @@ class MockedCubeService: cubes = ["Cube1", "Cube2"] - def get_all_names(self, cube_name: str): + def get_all_names(self, skip_control_cubes: bool = False): return self.cubes def exists(self, cube_name: str): @@ -14,7 +14,7 @@ def exists(self, cube_name: str): class MockedViewService: def get_all_names(self, cube_name: str): - return ["View1", "View2", "View3"] + return ([], ["View1", "View2", "View3"]) def exists(self, cube_name: str, view_name: str, private: bool): if "not" in view_name.lower(): @@ -35,7 +35,7 @@ def exists(self, dimension_name: str): class MockedSubsetService: - def get_all_names(self, dimension_name: str): + def get_all_names(self, dimension_name: str, hierarchy_name: str = None, private: bool = False): return ["Subset1", "Subset2", "Subset3"] def exists(self, dimension_name: str, subset_name: str, private: bool): @@ -46,6 +46,11 @@ def exists(self, dimension_name: str, subset_name: str, private: bool): class MockedProcessService: + processes = ["Process1", "Process2"] + + def get_all_names(self, skip_control_processes: bool = False): + return self.processes + def exists(self, process_name: str): return False if "not" in process_name else True diff --git a/tests/test_cmd_cubes.py b/tests/test_cmd_cubes.py index ed95515..05e8ac2 100644 --- a/tests/test_cmd_cubes.py +++ b/tests/test_cmd_cubes.py @@ -13,7 +13,7 @@ def test_cube_list(mocker, command): result = runner.invoke(app, ["cube", command]) assert result.exit_code == 0 assert isinstance(result.stdout, str) - assert result.stdout == "Cube1\nCube2\n" + assert result.stdout == "- Cube1\n- Cube2\n" def test_cube_exists(mocker): diff --git a/tests/test_cmd_dimension.py b/tests/test_cmd_dimension.py index e27e379..d6af33c 100644 --- a/tests/test_cmd_dimension.py +++ b/tests/test_cmd_dimension.py @@ -13,7 +13,7 @@ def test_dimension_list(mocker, command): result = runner.invoke(app, ["dimension", command]) assert result.exit_code == 0 assert isinstance(result.stdout, str) - assert result.stdout == "Dimension1\nDimension2\nDimension3\n" + assert result.stdout == "- Dimension1\n- Dimension2\n- Dimension3\n" def test_dimension_exists(mocker): diff --git a/tests/test_cmd_subset.py b/tests/test_cmd_subset.py index 3d6340d..8afdefa 100644 --- a/tests/test_cmd_subset.py +++ b/tests/test_cmd_subset.py @@ -10,10 +10,12 @@ @pytest.mark.parametrize("command", ["list", "ls"]) def test_subset_list(mocker, command): mocker.patch("tm1cli.commands.subset.TM1Service", MockedTM1Service) - result = runner.invoke(app, ["subset", command, "Dimension1"]) + result = runner.invoke(app, ["subset", command, "--dimension", "Dimension1"]) assert result.exit_code == 0 assert isinstance(result.stdout, str) - assert result.stdout == "Subset1\nSubset2\nSubset3\n" + assert "dimension: Dimension1" in result.stdout + assert "name: Subset1" in result.stdout + assert "type: public" in result.stdout def test_subset_exists(mocker): diff --git a/tests/test_cmd_view.py b/tests/test_cmd_view.py index 104ded4..b46c851 100644 --- a/tests/test_cmd_view.py +++ b/tests/test_cmd_view.py @@ -25,8 +25,10 @@ def test_view_exists(mocker, options): def test_view_list(mocker): mocker.patch("tm1cli.commands.view.TM1Service", MockedTM1Service) - result = runner.invoke(app, ["view", "list", "example_cube"]) + result = runner.invoke(app, ["view", "list", "--cube", "example_cube"]) assert result.exit_code == 0 assert isinstance(result.stdout, str) - assert result.stdout == "View1\nView2\nView3\n" + assert "cube: example_cube" in result.stdout + assert "name: View1" in result.stdout + assert "type: public" in result.stdout diff --git a/tm1cli/commands/cube.py b/tm1cli/commands/cube.py index ab88047..f3c688f 100644 --- a/tm1cli/commands/cube.py +++ b/tm1cli/commands/cube.py @@ -1,10 +1,11 @@ -from typing import Annotated +from typing import Annotated, Optional import typer from rich import print # pylint: disable=redefined-builtin from TM1py.Services import TM1Service from tm1cli.utils.cli_param import DATABASE_OPTION, INTERVAL_OPTION, WATCH_OPTION +from tm1cli.utils.list_utils import OutputFormat, apply_filter, apply_paging, render_output from tm1cli.utils.various import resolve_database from tm1cli.utils.watch import watch_option @@ -21,17 +22,22 @@ def list_cube( typer.Option( "-s", "--skip-control-cubes", - help="Flag for not printing control cubes.", + help="Exclude control cubes (names starting with '}').", ), ] = False, + filter: Annotated[Optional[str], typer.Option("--filter", "-f", help="Regex/substring filter (case-insensitive).")] = None, + output: Annotated[OutputFormat, typer.Option("--output", "-o", help="Output format: yaml (default) or json.")] = "yaml", + limit: Annotated[Optional[int], typer.Option("--limit", help="Maximum number of results to return.")] = None, + offset: Annotated[int, typer.Option("--offset", help="Number of results to skip (for paging).", min=0)] = 0, ): """ List cubes """ - with TM1Service(**resolve_database(ctx, database)) as tm1: - for cube in tm1.cubes.get_all_names(skip_control_cubes): - print(cube) + names = tm1.cubes.get_all_names(skip_control_cubes=skip_control_cubes) + names = apply_filter(names, filter) + result = apply_paging(names, limit, offset) + typer.echo(render_output(result, output)) @app.command() diff --git a/tm1cli/commands/dimension.py b/tm1cli/commands/dimension.py index 9926105..fc0b4f8 100644 --- a/tm1cli/commands/dimension.py +++ b/tm1cli/commands/dimension.py @@ -1,10 +1,11 @@ -from typing import Annotated +from typing import Annotated, Optional import typer from rich import print # pylint: disable=redefined-builtin from TM1py.Services import TM1Service from tm1cli.utils.cli_param import DATABASE_OPTION, INTERVAL_OPTION, WATCH_OPTION +from tm1cli.utils.list_utils import OutputFormat, apply_filter, apply_paging, render_output from tm1cli.utils.various import resolve_database from tm1cli.utils.watch import watch_option @@ -20,18 +21,23 @@ def list_dimension( bool, typer.Option( "-s", - "--skip-control-cubes", - help="Flag for not printing control cubes.", + "--skip-control-dims", + help="Exclude control dimensions (names starting with '}').", ), ] = False, + filter: Annotated[Optional[str], typer.Option("--filter", "-f", help="Regex/substring filter (case-insensitive).")] = None, + output: Annotated[OutputFormat, typer.Option("--output", "-o", help="Output format: yaml (default) or json.")] = "yaml", + limit: Annotated[Optional[int], typer.Option("--limit", help="Maximum number of results to return.")] = None, + offset: Annotated[int, typer.Option("--offset", help="Number of results to skip (for paging).", min=0)] = 0, ): """ List dimensions """ - with TM1Service(**resolve_database(ctx, database)) as tm1: - for dim in tm1.dimensions.get_all_names(skip_control_dims): - print(dim) + names = tm1.dimensions.get_all_names(skip_control_dims=skip_control_dims) + names = apply_filter(names, filter) + result = apply_paging(names, limit, offset) + typer.echo(render_output(result, output)) @app.command() diff --git a/tm1cli/commands/process.py b/tm1cli/commands/process.py index 7e1d46f..808610f 100644 --- a/tm1cli/commands/process.py +++ b/tm1cli/commands/process.py @@ -1,5 +1,6 @@ import json from pathlib import Path +from typing import Annotated, Optional import typer from rich import print # pylint: disable=redefined-builtin @@ -8,6 +9,7 @@ from typing_extensions import Annotated from tm1cli.utils.cli_param import DATABASE_OPTION, INTERVAL_OPTION, WATCH_OPTION +from tm1cli.utils.list_utils import OutputFormat, apply_filter, apply_paging, render_output from tm1cli.utils.tm1yaml import dump_process, load_process from tm1cli.utils.various import print_error_and_exit, resolve_database from tm1cli.utils.watch import watch_option @@ -28,14 +30,27 @@ def _get_process(name: str, database_config: dict) -> Process: def list_process( ctx: typer.Context, database: Annotated[str, DATABASE_OPTION] = None, + skip_control_tis: Annotated[ + bool, + typer.Option( + "-s", + "--skip-control-tis", + help="Exclude control TI processes (names starting with '}').", + ), + ] = False, + filter: Annotated[Optional[str], typer.Option("--filter", "-f", help="Regex/substring filter (case-insensitive).")] = None, + output: Annotated[OutputFormat, typer.Option("--output", "-o", help="Output format: yaml (default) or json.")] = "yaml", + limit: Annotated[Optional[int], typer.Option("--limit", help="Maximum number of results to return.")] = None, + offset: Annotated[int, typer.Option("--offset", help="Number of results to skip (for paging).", min=0)] = 0, ): """ List processes """ - with TM1Service(**resolve_database(ctx, database)) as tm1: - for process in tm1.processes.get_all_names(): - print(process) + names = tm1.processes.get_all_names(skip_control_processes=skip_control_tis) + names = apply_filter(names, filter) + result = apply_paging(names, limit, offset) + typer.echo(render_output(result, output)) @app.command() diff --git a/tm1cli/commands/subset.py b/tm1cli/commands/subset.py index 021e720..c436759 100644 --- a/tm1cli/commands/subset.py +++ b/tm1cli/commands/subset.py @@ -1,10 +1,11 @@ -from typing import Annotated +from typing import Annotated, Optional import typer from rich import print # pylint: disable=redefined-builtin from TM1py.Services import TM1Service from tm1cli.utils.cli_param import DATABASE_OPTION, INTERVAL_OPTION, WATCH_OPTION +from tm1cli.utils.list_utils import OutputFormat, VisibilityType, apply_filter, apply_paging, render_output from tm1cli.utils.various import resolve_database from tm1cli.utils.watch import watch_option @@ -15,17 +16,44 @@ @app.command(name="list") def list_subset( ctx: typer.Context, - dimension_name: str, - # hierarchy_name: str = None, + dimension: Annotated[ + Optional[str], + typer.Option("--dimension", help="Scope to a specific dimension name."), + ] = None, + hierarchy: Annotated[ + Optional[str], + typer.Option("--hierarchy", help="Hierarchy name (defaults to dimension name)."), + ] = None, + type: Annotated[ + VisibilityType, + typer.Option("--type", "-t", help="Visibility filter: public, private, or both."), + ] = "public", database: Annotated[str, DATABASE_OPTION] = None, + filter: Annotated[Optional[str], typer.Option("--filter", "-f", help="Regex/substring filter (case-insensitive).")] = None, + output: Annotated[OutputFormat, typer.Option("--output", "-o", help="Output format: yaml (default) or json.")] = "yaml", + limit: Annotated[Optional[int], typer.Option("--limit", help="Maximum number of results to return.")] = None, + offset: Annotated[int, typer.Option("--offset", help="Number of results to skip (for paging).", min=0)] = 0, ): """ List subsets """ - + include_public = type in ("public", "both") + include_private = type in ("private", "both") with TM1Service(**resolve_database(ctx, database)) as tm1: - for subset in tm1.subsets.get_all_names(dimension_name): - print(subset) + dim_names = [dimension] if dimension else tm1.dimensions.get_all_names(skip_control_dims=True) + results: list[dict] = [] + for dim in dim_names: + hier = hierarchy or dim + if include_public: + for name in tm1.subsets.get_all_names(dim, hier, private=False): + results.append({"dimension": dim, "hierarchy": hier, "name": name, "type": "public"}) + if include_private: + for name in tm1.subsets.get_all_names(dim, hier, private=True): + results.append({"dimension": dim, "hierarchy": hier, "name": name, "type": "private"}) + if filter: + matching = set(apply_filter([r["name"] for r in results], filter)) + results = [r for r in results if r["name"] in matching] + typer.echo(render_output(apply_paging(results, limit, offset), output)) @app.command() diff --git a/tm1cli/commands/view.py b/tm1cli/commands/view.py index d71768d..f68ff2f 100644 --- a/tm1cli/commands/view.py +++ b/tm1cli/commands/view.py @@ -1,10 +1,11 @@ -from typing import Annotated +from typing import Annotated, Optional import typer from rich import print # pylint: disable=redefined-builtin from TM1py.Services import TM1Service from tm1cli.utils.cli_param import DATABASE_OPTION, INTERVAL_OPTION, WATCH_OPTION +from tm1cli.utils.list_utils import OutputFormat, VisibilityType, apply_filter, apply_paging, render_output from tm1cli.utils.various import resolve_database from tm1cli.utils.watch import watch_option @@ -15,16 +16,37 @@ @app.command(name="list") def list_view( ctx: typer.Context, - cube_name: str, + cube: Annotated[Optional[str], typer.Option("--cube", "-c", help="Scope to a specific cube name.")] = None, + type: Annotated[ + VisibilityType, + typer.Option("--type", "-t", help="Visibility filter: public, private, or both."), + ] = "public", database: Annotated[str, DATABASE_OPTION] = None, + filter: Annotated[Optional[str], typer.Option("--filter", "-f", help="Regex/substring filter (case-insensitive).")] = None, + output: Annotated[OutputFormat, typer.Option("--output", "-o", help="Output format: yaml (default) or json.")] = "yaml", + limit: Annotated[Optional[int], typer.Option("--limit", help="Maximum number of results to return.")] = None, + offset: Annotated[int, typer.Option("--offset", help="Number of results to skip (for paging).", min=0)] = 0, ): """ List views """ - + include_public = type in ("public", "both") + include_private = type in ("private", "both") with TM1Service(**resolve_database(ctx, database)) as tm1: - for view in tm1.views.get_all_names(cube_name): - print(view) + cube_names = [cube] if cube else tm1.cubes.get_all_names(skip_control_cubes=True) + results: list[dict] = [] + for cube_name in cube_names: + private_names, public_names = tm1.views.get_all_names(cube_name) + if include_public: + for name in public_names: + results.append({"cube": cube_name, "name": name, "type": "public"}) + if include_private: + for name in private_names: + results.append({"cube": cube_name, "name": name, "type": "private"}) + if filter: + matching = set(apply_filter([r["name"] for r in results], filter)) + results = [r for r in results if r["name"] in matching] + typer.echo(render_output(apply_paging(results, limit, offset), output)) @app.command() diff --git a/tm1cli/utils/list_utils.py b/tm1cli/utils/list_utils.py new file mode 100644 index 0000000..24a6c01 --- /dev/null +++ b/tm1cli/utils/list_utils.py @@ -0,0 +1,48 @@ +"""Shared utilities for list commands: filtering, paging, and output rendering.""" +from __future__ import annotations + +import json +import re +from enum import Enum +from typing import Optional + +import yaml + + +class OutputFormat(str, Enum): + yaml = "yaml" + json = "json" + + +class VisibilityType(str, Enum): + public = "public" + private = "private" + both = "both" + + +def apply_filter(names: list[str], pattern: Optional[str]) -> list[str]: + """Return names matching *pattern* (case-insensitive regex or substring).""" + if not pattern: + return names + try: + rx = re.compile(pattern, re.IGNORECASE) + return [n for n in names if rx.search(n)] + except re.error: + needle = pattern.lower() + return [n for n in names if needle in n.lower()] + + +def apply_paging(items: list, limit: Optional[int], offset: int) -> list: + """Slice *items* by *offset* and *limit*.""" + if offset: + items = items[offset:] + if limit is not None: + items = items[:limit] + return items + + +def render_output(data: list, fmt: OutputFormat) -> str: + """Serialise *data* to a YAML or JSON string.""" + if fmt == "json": + return json.dumps(data, indent=2, ensure_ascii=False) + return yaml.dump(data, default_flow_style=False, allow_unicode=True, sort_keys=False).rstrip() From 0f3aa26716bb609167f7c926fcd74878607dfa4f Mon Sep 17 00:00:00 2001 From: Enis Isufi Date: Fri, 14 Aug 2026 09:53:00 +0200 Subject: [PATCH 2/3] Added test_cmd_process.py and minor fixes --- tests/test_cmd_process.py | 62 ++++++++++++++++++++++++++++++++++++++ tests/test_tm1cli.py | 10 ++++-- tm1cli/commands/process.py | 1 - tm1cli/main.py | 11 ++++--- 4 files changed, 77 insertions(+), 7 deletions(-) create mode 100644 tests/test_cmd_process.py diff --git a/tests/test_cmd_process.py b/tests/test_cmd_process.py new file mode 100644 index 0000000..258c7df --- /dev/null +++ b/tests/test_cmd_process.py @@ -0,0 +1,62 @@ +import pytest +from typer.testing import CliRunner + +from tests.conftest import MockedTM1Service +from tm1cli.main import app + +runner = CliRunner() + + +@pytest.mark.parametrize("command", ["list", "ls"]) +def test_process_list(mocker, command): + mocker.patch("tm1cli.commands.process.TM1Service", MockedTM1Service) + result = runner.invoke(app, ["process", command]) + assert result.exit_code == 0 + assert isinstance(result.stdout, str) + assert result.stdout == "- Process1\n- Process2\n" + + +def test_process_list_json(mocker): + mocker.patch("tm1cli.commands.process.TM1Service", MockedTM1Service) + result = runner.invoke(app, ["process", "list", "--output", "json"]) + assert result.exit_code == 0 + assert '"Process1"' in result.stdout + assert '"Process2"' in result.stdout + + +def test_process_list_filter(mocker): + mocker.patch("tm1cli.commands.process.TM1Service", MockedTM1Service) + result = runner.invoke(app, ["process", "list", "--filter", "Process1"]) + assert result.exit_code == 0 + assert "Process1" in result.stdout + assert "Process2" not in result.stdout + + +def test_process_list_limit(mocker): + mocker.patch("tm1cli.commands.process.TM1Service", MockedTM1Service) + result = runner.invoke(app, ["process", "list", "--limit", "1"]) + assert result.exit_code == 0 + assert "Process1" in result.stdout + assert "Process2" not in result.stdout + + +def test_process_list_offset(mocker): + mocker.patch("tm1cli.commands.process.TM1Service", MockedTM1Service) + result = runner.invoke(app, ["process", "list", "--offset", "1"]) + assert result.exit_code == 0 + assert "Process1" not in result.stdout + assert "Process2" in result.stdout + + +def test_process_exists(mocker): + mocker.patch("tm1cli.commands.process.TM1Service", MockedTM1Service) + result = runner.invoke(app, ["process", "exists", "Process1"]) + assert result.exit_code == 0 + assert result.stdout == "True\n" + + +def test_process_not_exists(mocker): + mocker.patch("tm1cli.commands.process.TM1Service", MockedTM1Service) + result = runner.invoke(app, ["process", "exists", "process_not_found"]) + assert result.exit_code == 0 + assert result.stdout == "False\n" diff --git a/tests/test_tm1cli.py b/tests/test_tm1cli.py index 9d5e538..0b458b5 100644 --- a/tests/test_tm1cli.py +++ b/tests/test_tm1cli.py @@ -1,6 +1,7 @@ import pytest from typer.testing import CliRunner +from tests.conftest import MockedTM1Service from tm1cli.main import app runner = CliRunner() @@ -59,8 +60,13 @@ def test_process_clone_missing_from_to(): ) -def test_process_clone_not_exists(): - result = runner.invoke(app, ["process", "clone", "example", "--to", "remotedb"]) +def test_process_clone_not_exists(mocker): + mocker.patch("tm1cli.commands.process.TM1Service", MockedTM1Service) + mocker.patch( + "tm1cli.commands.process.resolve_database", + side_effect=lambda ctx, db: {"address": "target", "port": 8080} if db == "remotedb" else {"address": "source", "port": 8005}, + ) + result = runner.invoke(app, ["process", "clone", "process_notfound", "--to", "remotedb"]) assert result.exit_code == 1 assert "Error: Process does not exist in source database!" in result.output diff --git a/tm1cli/commands/process.py b/tm1cli/commands/process.py index 808610f..9f0cceb 100644 --- a/tm1cli/commands/process.py +++ b/tm1cli/commands/process.py @@ -6,7 +6,6 @@ from rich import print # pylint: disable=redefined-builtin from TM1py.Objects import Process from TM1py.Services import TM1Service -from typing_extensions import Annotated from tm1cli.utils.cli_param import DATABASE_OPTION, INTERVAL_OPTION, WATCH_OPTION from tm1cli.utils.list_utils import OutputFormat, apply_filter, apply_paging, render_output diff --git a/tm1cli/main.py b/tm1cli/main.py index e7a3b02..f2afba1 100644 --- a/tm1cli/main.py +++ b/tm1cli/main.py @@ -99,10 +99,13 @@ def threads( with TM1Service(**db_config) as tm1: threads = tm1.sessions.get_threads_for_current() if beautify: - table = Table(*threads[0].keys(), title="Threads") - for thread in threads: - table.add_row(*[str(value) for value in thread.values()]) - console.print(table) + if not threads: + typer.echo("No threads.") + else: + table = Table(*threads[0].keys(), title="Threads") + for thread in threads: + table.add_row(*[str(value) for value in thread.values()]) + console.print(table) else: threads = json.dumps(threads, indent=4) print(threads) From 27e74038da90ad612d3fb5555de840329d527428 Mon Sep 17 00:00:00 2001 From: Enis Isufi Date: Fri, 14 Aug 2026 11:14:57 +0200 Subject: [PATCH 3/3] reversed unwanted change --- tests/test_tm1cli.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/test_tm1cli.py b/tests/test_tm1cli.py index 2bc2021..c0092c0 100644 --- a/tests/test_tm1cli.py +++ b/tests/test_tm1cli.py @@ -7,7 +7,7 @@ runner = CliRunner() -@pytest.mark.parametrize("option", [None, ["--database", "remotedb"], ["-d", "remotedb"]]) +@pytest.mark.parametrize("option", [None, ["--database", "mydb"], ["-d", "remotedb"]]) def test_tm1_version(option): if option: result = runner.invoke(app, ["tm1-version", *option])