diff --git a/owilix/app/appkernel.py b/owilix/app/appkernel.py index 7faae09..7932b05 100644 --- a/owilix/app/appkernel.py +++ b/owilix/app/appkernel.py @@ -7,9 +7,13 @@ from rich.theme import Theme from owilix.core import OWIlixManager from owilix.core.manager import OWIlixConfig, OWILIXEnv -from owilix.cmd import LocalCommands, AdminCommands, RemoteCommands, ConfigCommands -from owilix.cmd.query import QueryCommands -from owilix.cmd.workflows import WorkflowCommands +from owilix.compat import LocalCommands, ConfigCommands +from owilix.compat.workflows import WorkflowCommands +# TODO: AdminCommands, RemoteCommands, QueryCommands were never implemented +# in the legacy cmd system. Stub them out until the server app is migrated. +AdminCommands = None +RemoteCommands = None +QueryCommands = None from .console_stream import StreamingConsole @@ -76,10 +80,7 @@ class AppKernel: # Register command groups with this ctx self.registry = { "local": LocalCommands(owi, **self.ctx), - "remote": RemoteCommands(owi, **self.ctx), - "admin": AdminCommands(owi, **self.ctx), "config": ConfigCommands(owi, **self.ctx), - "query": QueryCommands(owi, **self.ctx), "workflows": WorkflowCommands(owi, **self.ctx), } diff --git a/owilix/cli/__init__.py b/owilix/cli/__init__.py index 63a5379..95a5317 100644 --- a/owilix/cli/__init__.py +++ b/owilix/cli/__init__.py @@ -105,11 +105,11 @@ def main( no_remote_config=no_remote_config, ) - # Patch legacy progress bars if globally suppressed + # Suppress progress bars globally when --no-progress is set if no_progress: - import owilix.cmd.base + import owilix.core.manager.ui from ._common.progress import DummyProgress - owilix.cmd.base.currentItemProgress = lambda *args, **kwargs: DummyProgress() + owilix.core.manager.ui.currentItemProgress = lambda *args, **kwargs: DummyProgress() # Import and register sub-apps diff --git a/owilix/cli/_common/ui.py b/owilix/cli/_common/ui.py index 1193135..28c5135 100644 --- a/owilix/cli/_common/ui.py +++ b/owilix/cli/_common/ui.py @@ -2,6 +2,7 @@ from owilix.core.manager.ui import ( currentItemProgress, ask_yes_no, input_dict, + render_table_dynamic, EnhancedProgressDisplay, ErrorCollector, ErrorCategory, diff --git a/owilix/cli/batch.py b/owilix/cli/batch.py index f15874f..6e2c5ed 100644 --- a/owilix/cli/batch.py +++ b/owilix/cli/batch.py @@ -49,7 +49,7 @@ def batch_run( raise typer.Exit(code=1) # Delegate to legacy BatchCommands - from owilix.cmd.batch import BatchCommands + from owilix.compat.batch import BatchCommands batch_cmd = BatchCommands(cli_ctx.owi, console=cli_ctx.console) diff --git a/owilix/cli/local.py b/owilix/cli/local.py index 2a8952b..fec18bf 100644 --- a/owilix/cli/local.py +++ b/owilix/cli/local.py @@ -227,7 +227,7 @@ def analyze( cli_ctx: CLIContext = ctx.obj # Use the legacy implementation for complex analysis - from owilix.cmd.local import LocalCommands + from owilix.compat.local import LocalCommands local_cmd = LocalCommands( cli_ctx.owi, diff --git a/owilix/cmd/__init__.py b/owilix/cmd/__init__.py deleted file mode 100644 index 8d83650..0000000 --- a/owilix/cmd/__init__.py +++ /dev/null @@ -1,17 +0,0 @@ -""" -Package containing the different commands for OWIlix. - -Commands are intended to be run via the CLI, receive context objects from the CLI and can interact with the user while -using the owilix.core API, whereas the rich cmd python library is used for interaction. - -`owilix.cmd.base.BaseCommands` defines a base class for implementing commands. It thereby defines a register method, -that allows to register methods as commands for this class. - - -""" - -from .local import LocalCommands - -from .config import ConfigCommands - -from .batch import BatchCommands \ No newline at end of file diff --git a/owilix/cmd/base.py b/owilix/cmd/base.py deleted file mode 100644 index 5b33cc7..0000000 --- a/owilix/cmd/base.py +++ /dev/null @@ -1,17 +0,0 @@ -""" -Base Command Module (Legacy/Aggregation) - -This module aggregates the base command functionality now split across: -- owilix.cmd.core_base (BaseCommand, SubCommand, etc.) -- owilix.cmd.sql_base (SQLBaseCommands) - -It is maintained for backward compatibility. -""" - -from .core_base import ( - BaseCommand, SubCommand, SubCommandMeta, CommandResult, - get_func_params, show_dataframe, render_dataframe, render_table, - render_table_dynamic, currentItemProgress, ask_yes_no, input_dict -) - - diff --git a/owilix/compat/__init__.py b/owilix/compat/__init__.py new file mode 100644 index 0000000..9b21ee3 --- /dev/null +++ b/owilix/compat/__init__.py @@ -0,0 +1,20 @@ +""" +Backward-compatibility shim for the legacy Click-based command system. + +The active CLI is Typer-based in ``owilix.cli``. This package preserves +``BaseCommand``, ``SubCommand``, and the command classes (``LocalCommands``, +``ConfigCommands``, ``BatchCommands``, ``WorkflowCommands``) so that +plugins and the server app-kernel can still import them. + +New code should NOT import from here — use ``owilix.core.types`` for +``CommandResult`` and ``owilix.core.manager.ui`` for UI utilities. +""" + +from .core_base import BaseCommand, SubCommand, SubCommandMeta, get_func_params +from .local import LocalCommands +from .config import ConfigCommands +from .batch import BatchCommands + +# Re-export commonly used types so plugins can import from owilix.compat +from owilix.core.types import CommandResult, ErrorType, ExitCode +from owilix.core.manager.ui import currentItemProgress, ask_yes_no, input_dict, render_table_dynamic diff --git a/owilix/compat/base.py b/owilix/compat/base.py new file mode 100644 index 0000000..a2eb7b1 --- /dev/null +++ b/owilix/compat/base.py @@ -0,0 +1,17 @@ +""" +Base Command Module (Legacy/Aggregation) + +This module aggregates the base command functionality. +Maintained for backward compatibility — new code should import +from ``owilix.core.types`` and ``owilix.core.manager.ui`` directly. +""" + +from .core_base import ( + BaseCommand, SubCommand, SubCommandMeta, + get_func_params, show_dataframe, render_dataframe, render_table, +) + +from owilix.core.types import CommandResult, ErrorType, ExitCode +from owilix.core.manager.ui import ( + render_table_dynamic, currentItemProgress, ask_yes_no, input_dict +) diff --git a/owilix/cmd/batch.py b/owilix/compat/batch.py similarity index 100% rename from owilix/cmd/batch.py rename to owilix/compat/batch.py diff --git a/owilix/cmd/config.py b/owilix/compat/config.py similarity index 98% rename from owilix/cmd/config.py rename to owilix/compat/config.py index 4ae9c50..42e3a52 100644 --- a/owilix/cmd/config.py +++ b/owilix/compat/config.py @@ -13,7 +13,8 @@ Provides commands for viewing, editing, and migrating configuration: import os -from owilix.cmd.base import BaseCommand, SubCommand, CommandResult +from owilix.compat.base import BaseCommand, SubCommand +from owilix.core.types import CommandResult from owilix.core.utils import dict_multi_key_get, dict_multi_key_add, dict_multi_key_delete diff --git a/owilix/cmd/core_base.py b/owilix/compat/core_base.py similarity index 100% rename from owilix/cmd/core_base.py rename to owilix/compat/core_base.py diff --git a/owilix/cmd/local.py b/owilix/compat/local.py similarity index 99% rename from owilix/cmd/local.py rename to owilix/compat/local.py index cd6829d..ed26a4b 100644 --- a/owilix/cmd/local.py +++ b/owilix/compat/local.py @@ -4,7 +4,9 @@ import traceback from rich.prompt import Prompt import owilix -from owilix.cmd.base import BaseCommand, input_dict, SubCommand, currentItemProgress, CommandResult, ask_yes_no +from owilix.compat.base import BaseCommand, SubCommand +from owilix.core.types import CommandResult +from owilix.core.manager.ui import input_dict, currentItemProgress, ask_yes_no from owilix.core.models.dataset import infer_metadata_from_files from owilix.core.utils import get_filesystem, fill_file_details, compare_filesystems, \ group_and_count_files, rel_can_path diff --git a/owilix/cmd/subcmds/__init__.py b/owilix/compat/subcmds/__init__.py similarity index 100% rename from owilix/cmd/subcmds/__init__.py rename to owilix/compat/subcmds/__init__.py diff --git a/owilix/cmd/subcmds/graph_utils.py b/owilix/compat/subcmds/graph_utils.py similarity index 100% rename from owilix/cmd/subcmds/graph_utils.py rename to owilix/compat/subcmds/graph_utils.py diff --git a/owilix/cmd/workflows.py b/owilix/compat/workflows.py similarity index 99% rename from owilix/cmd/workflows.py rename to owilix/compat/workflows.py index abd1137..4bec5ea 100644 --- a/owilix/cmd/workflows.py +++ b/owilix/compat/workflows.py @@ -38,7 +38,8 @@ from statistics import mean from typing import Any, Dict, List, Optional import os, json, requests -from owilix.cmd.base import CommandResult, SubCommand, BaseCommand +from owilix.compat.base import BaseCommand, SubCommand +from owilix.core.types import CommandResult # -------- Airflow monitor wrapper (no top-level /dagRuns calls) -------- diff --git a/owilix/core/manager/ui.py b/owilix/core/manager/ui.py index 229493e..eb6f6b6 100644 --- a/owilix/core/manager/ui.py +++ b/owilix/core/manager/ui.py @@ -1,5 +1,6 @@ import json import logging +import re import threading import uuid from collections import deque, defaultdict @@ -13,6 +14,7 @@ from rich.console import Console, Group from rich.live import Live from rich.panel import Panel from rich.progress import BarColumn, Progress, TextColumn, TimeRemainingColumn +from rich import box from rich.table import Table from rich.text import Text @@ -864,3 +866,75 @@ def input_dict(data: dict, title="User Data", console=None, fn_valide=None): display_data(data) return data + +# --------------------------------------------------------------------------- +# Table rendering utilities (migrated from owilix.cmd.core_base) +# --------------------------------------------------------------------------- + +def render_table_dynamic(data: list, console: Console, show_header: bool = True, + max_cols: int = 4, ignore=(), colorder=(), roworder=(), + showfields=(), **kwargs): + """ + Render a list of dictionaries to the console as a table with dynamic column widths, + splitting rows if the number of columns exceeds the specified `max_cols`. + + Parameters: + data (list): The list of dictionaries to be rendered. + console (rich.console.Console): A Rich console object used to display the table. + show_header (bool, optional): If True, display the table headers. Defaults to True. + max_cols (int, optional): Maximum number of columns to display per row. Additional + columns are displayed as labels in separate rows. Defaults to 4. + ignore (tuple, optional): Regex patterns for columns to ignore. Defaults to (). + colorder (tuple, optional): Desired column order. Defaults to (). + roworder (tuple, optional): Sort order for rows, with keys and descending flags. Defaults to (). + showfields (tuple, optional): Columns to always include, even if ignored. Defaults to (). + **kwargs: Additional keyword arguments to pass to the Rich Table constructor. + """ + if not data: + console.print("No data available to display.") + return + + table = Table(show_header=show_header, box=box.SIMPLE, **kwargs) + + # Determine the columns from the keys of the first dictionary + _mcols = set([c for d in data for c in d.keys() if not any([re.match(i, c) for i in ignore]) or c in showfields]) + columns = list([c for c in colorder if c in set(colorder).intersection(_mcols)]) + list(_mcols.difference(colorder)) + + if roworder and len(roworder) > 0: + # Sort the data based on the row order + for descending, key in reversed(roworder): + data = sorted(data, key=lambda x: (x.get(key) is None, x.get(key)), reverse=descending) + + # Split columns into chunks + column_chunks = [columns[i:i + max_cols] for i in range(0, len(columns), max_cols)] + + # Add headers for the first chunk + for column in column_chunks[0]: + table.add_column(column) + + # Add rows + for row in data: + # Add the first chunk as the main row + main_row_values = [str(row.get(column, "")) for column in column_chunks[0]] + table.add_row(*main_row_values) + + # Add additional chunks as indented rows + for chunk in column_chunks[1:]: + chunk_values = [] + for column in chunk: + value = column + ":" + str(row.get(column, "")) + max_width = console.width // len(chunk) # Dynamic width for each column in the chunk + # Split long text into chunks to fit in columns + lines = [value[i:i + max_width] for i in range(0, len(value), max_width)] + chunk_values.extend(lines) + + # Pad with blanks for remaining columns in the chunk + while len(chunk_values) < len(chunk): + chunk_values.append("") + + indented_row = [" "] + chunk_values[:len(chunk)] # Indent for continuation rows + table.add_row(*indented_row) + table.add_section() + + console.print(table) + diff --git a/owilix/core/tasks/_graph_utils.py b/owilix/core/tasks/_graph_utils.py new file mode 100644 index 0000000..be3257d --- /dev/null +++ b/owilix/core/tasks/_graph_utils.py @@ -0,0 +1,429 @@ +# =========================== +# rustworkx graph extra stats +# =========================== +from __future__ import annotations +from typing import Dict, List, Tuple, Iterable, Optional, Set, Any +import math +import random +from collections import deque, defaultdict, Counter + +import numpy as np +import pandas as pd +import rustworkx as rx + + +# ------------------------------------------------------------ +# 1) Approximate betweenness centrality (sampled Brandes) +# ------------------------------------------------------------ +def approximate_betweenness_centrality( + G: rx.PyDiGraph, + *, + k: int = 500, + seed: Optional[int] = None, + normalized: bool = True, + directed: bool = True, +) -> Dict[Any, float]: + """ + Approximate betweenness centrality by sampling k source nodes and + running a Brandes single-source dependency accumulation per sample. + + Args: + G: rustworkx PyDiGraph + k: number of source nodes to sample (<= |V|) + seed: RNG seed for reproducibility + normalized: if True, divide by (n-1)(n-2) [directed], or by ((n-1)(n-2)/2) [undirected] + directed: normalization mode + + Returns: + dict mapping node payload (labels) -> approx betweenness score + """ + n = G.num_nodes() + if n == 0: + return {} + indices = list(range(n)) + if seed is not None: + random.seed(seed) + + if k >= n: + sources = indices + else: + sources = random.sample(indices, k) + + # map idx->label for readable keys + idx_to_label = {i: G[i] for i in indices} + + Cb = defaultdict(float) + + # --- single-source Brandes (unweighted, directed) on PyDiGraph --- + # Uses BFS layer expansion via G.out_edges(u) to collect successors. + for s in sources: + S_stack: List[int] = [] + P: Dict[int, List[int]] = defaultdict(list) # predecessors on shortest paths + sigma: Dict[int, float] = defaultdict(float) # number of shortest paths + dist: Dict[int, int] = defaultdict(lambda: -1) + + sigma[s] = 1.0 + dist[s] = 0 + Q = deque([s]) + + # BFS to get shortest paths counts + while Q: + v = Q.popleft() + S_stack.append(v) + # iterate successors (out-neighbors) + for _, w, _ in G.out_edges(v): + if dist[w] < 0: + dist[w] = dist[v] + 1 + Q.append(w) + if dist[w] == dist[v] + 1: + sigma[w] += sigma[v] + P[w].append(v) + + # accumulation + delta: Dict[int, float] = defaultdict(float) + while S_stack: + w = S_stack.pop() + for v in P[w]: + if sigma[w] > 0: + delta_v = (sigma[v] / sigma[w]) * (1.0 + delta[w]) + delta[v] += delta_v + if w != s: + Cb[w] += delta[w] + + # normalize + if normalized and n > 2: + if directed: + norm = 1.0 / ((n - 1) * (n - 2)) + else: + norm = 2.0 / ((n - 1) * (n - 2)) + for w in list(Cb.keys()): + Cb[w] *= norm + + # return keyed by labels + return {idx_to_label[i]: float(val) for i, val in Cb.items()} + + +# ------------------------------------------------------------ +# 2) WCC/SCC + simple bow-tie decomposition +# ------------------------------------------------------------ +def wcc_scc_bow_tie( + G: rx.PyDiGraph, +) -> Dict[str, Any]: + """ + Compute weak/strong components and a simple bow-tie partition: + - SCC0: nodes of the largest SCC + - IN: can reach SCC0 but not in it + - OUT: reachable from SCC0 but not in it + - OTHER: remaining (tendrils, tubes, disconnected) + + Returns sizes and node sets. + """ + n = G.num_nodes() + if n == 0: + return { + "num_wcc": 0, "num_scc": 0, + "sizes_wcc": [], "sizes_scc": [], + "largest_scc_size": 0, + "bow_tie": {"IN": set(), "SCC": set(), "OUT": set(), "OTHER": set()}, + } + + # Weakly Connected Components (works on DiGraph) + try: + wccs = rx.weakly_connected_components(G) + except Exception: + # Fallback: treat each node alone + wccs = [{i} for i in range(n)] + sizes_wcc = sorted([len(c) for c in wccs], reverse=True) + + # Strongly Connected Components + sccs = rx.strongly_connected_components(G) + sizes_scc = sorted([len(c) for c in sccs], reverse=True) + largest_scc = max(sccs, key=len) if sccs else set() + + # Helper: multi-source reachability + def forward_reachable(seeds: Set[int]) -> Set[int]: + seen: Set[int] = set() + dq = deque(seeds) + for s in seeds: + seen.add(s) + while dq: + v = dq.popleft() + for _, w, _ in G.out_edges(v): + if w not in seen: + seen.add(w) + dq.append(w) + return seen + + def backward_reachable(seeds: Set[int]) -> Set[int]: + seen: Set[int] = set() + dq = deque(seeds) + for s in seeds: + seen.add(s) + while dq: + v = dq.popleft() + for _, u, _ in G.in_edges(v): + if u not in seen: + seen.add(u) + dq.append(u) + return seen + + SCC = set(largest_scc) + IN = backward_reachable(SCC) - SCC + OUT = forward_reachable(SCC) - SCC + OTHER = set(range(n)) - SCC - IN - OUT + + idx_to_label = {i: G[i] for i in range(n)} + + return { + "num_wcc": len(wccs), + "num_scc": len(sccs), + "sizes_wcc": sizes_wcc, + "sizes_scc": sizes_scc, + "largest_scc_size": len(SCC), + "bow_tie": { + "IN": {idx_to_label[i] for i in IN}, + "SCC": {idx_to_label[i] for i in SCC}, + "OUT": {idx_to_label[i] for i in OUT}, + "OTHER": {idx_to_label[i] for i in OTHER}, + } + } + + +# ------------------------------------------------------------ +# 3) Reciprocity +# ------------------------------------------------------------ +def reciprocity(G: rx.PyDiGraph) -> Dict[str, float]: + """ + Global reciprocity: fraction of edges whose reverse also exists. + """ + m = G.num_edges() + if m == 0: + return {"reciprocity": 0.0, "mutual_pairs": 0, "edges": 0} + + # Build a hash set of directed edge endpoints (u,v) + edges = set() + for u, v in G.edge_list(): + if u != v: + edges.add((u, v)) + + mutual = 0 + seen_pairs = set() + for (u, v) in edges: + if (v, u) in edges and (v, u) not in seen_pairs: + mutual += 1 + seen_pairs.add((u, v)) + seen_pairs.add((v, u)) + + # Each mutual pair contributes 2 directed edges; ratio is (2*mutual)/m + rec = (2.0 * mutual) / float(m) if m > 0 else 0.0 + return {"reciprocity": rec, "mutual_pairs": mutual, "edges": float(m)} + + +# ------------------------------------------------------------ +# 4) Assortativity (degree mixing, directed variants) +# ------------------------------------------------------------ +def assortativity_degrees(G: rx.PyDiGraph) -> Dict[str, float]: + """ + Degree assortativity variants: + - out→out: corr(out(u), out(v)) over edges (u->v) + - in→in: corr(in(u), in(v)) over edges + - out→in: corr(out(u), in(v)) over edges + """ + n = G.num_nodes() + if n == 0 or G.num_edges() == 0: + return {"assort_out_out": np.nan, "assort_in_in": np.nan, "assort_out_in": np.nan} + + outdeg = {i: G.out_degree(i) for i in range(n)} + indeg = {i: G.in_degree(i) for i in range(n)} + + xs, ys = [], [] + for u, v in G.edge_list(): + xs.append(outdeg[u]); ys.append(outdeg[v]) + assort_oo = float(np.corrcoef(xs, ys)[0, 1]) if len(xs) > 1 else np.nan + + xs, ys = [], [] + for u, v in G.edge_list(): + xs.append(indeg[u]); ys.append(indeg[v]) + assort_ii = float(np.corrcoef(xs, ys)[0, 1]) if len(xs) > 1 else np.nan + + xs, ys = [], [] + for u, v in G.edge_list(): + xs.append(outdeg[u]); ys.append(indeg[v]) + assort_oi = float(np.corrcoef(xs, ys)[0, 1]) if len(xs) > 1 else np.nan + + return { + "assort_out_out": assort_oo, + "assort_in_in": assort_ii, + "assort_out_in": assort_oi, + } + + +# ------------------------------------------------------------ +# 5) Inequality: Gini + optional Lorenz curve points +# ------------------------------------------------------------ +def gini_and_lorenz(values: Iterable[float], lorenz_points: int = 200) -> Dict[str, Any]: + """ + Compute Gini coefficient and an optional Lorenz curve sample. + """ + arr = np.array([float(x) for x in values if x is not None], dtype=float) + if arr.size == 0: + return {"gini": np.nan, "lorenz": []} + + if np.any(arr < 0): + # shift if necessary (degenerately) + arr = arr - arr.min() + + if arr.sum() == 0: + return {"gini": 0.0, "lorenz": [(0.0, 0.0), (1.0, 1.0)]} + + arr = np.sort(arr) + n = arr.size + cum = np.cumsum(arr) + # Gini via relative mean difference + gini = (n + 1 - 2 * np.sum(cum) / cum[-1]) / n + + # Lorenz curve resampled to lorenz_points + x = np.linspace(0.0, 1.0, n, endpoint=True) + y = cum / cum[-1] + # resample + xi = np.linspace(0.0, 1.0, min(lorenz_points, n)) + yi = np.interp(xi, x, y) + lorenz = list(zip(xi.tolist(), yi.tolist())) + return {"gini": float(gini), "lorenz": lorenz} + + +# ------------------------------------------------------------ +# 6) Modularity (Louvain on undirected projection) +# ------------------------------------------------------------ +def undirected_projection(G: rx.PyDiGraph) -> Tuple[rx.PyGraph, Dict[int, int]]: + """ + Create an undirected PyGraph where each undirected edge weight is the sum of + forward and backward weights (or 1.0 if no weights). Also returns a mapping + from DiGraph node index -> PyGraph node index. + """ + UG = rx.PyGraph() + map_idx = {} + for i in range(G.num_nodes()): + map_idx[i] = UG.add_node(G[i]) + + # sum both directions for weight + seen = set() + for u, v in G.edge_list(): + a, b = (u, v) if u <= v else (v, u) + if (a, b) in seen or a == b: + continue + # sum weights (if present) + w_ab = 0.0 + for _, j, w in G.out_edges(a): + if j == b: + w_ab += float(w or 1.0) + for _, j, w in G.out_edges(b): + if j == a: + w_ab += float(w or 1.0) + if w_ab <= 0.0: + w_ab = 1.0 + UG.add_edge(map_idx[a], map_idx[b], w_ab) + seen.add((a, b)) + return UG, map_idx + + +# ------------------------------------------------------------ +# 7) Conductance for top communities (on undirected projection) +# ------------------------------------------------------------ +def conductance_for_partition( + G: rx.PyDiGraph, + communities: List[Set[int]], + top_m: int = 10 +) -> List[Dict[str, Any]]: + """ + Compute conductance φ(S) for top_m communities on UG. + φ(S) = cut(S,~S) / min(vol(S), vol(~S)), with volumes using weighted degrees. + """ + UG, _ = undirected_projection(G) + m = min(top_m, len(communities)) + results = [] + + # Precompute weighted degree for UG + wdeg = {i: 0.0 for i in range(UG.num_nodes())} + for i in range(UG.num_nodes()): + for _, j, w in UG.out_edges(i): + wdeg[i] += float(w or 1.0) + + # Build a quick edge lookup with weights + # UG.edge_list() returns pairs; need weights via weighted_edge_list() + cut_weights = defaultdict(dict) + for u, v, w in UG.weighted_edge_list(): + cut_weights[u][v] = float(w or 1.0) + cut_weights[v][u] = float(w or 1.0) + + for cid, comm in enumerate(communities[:m]): + S = set(comm) + Sc = set(range(UG.num_nodes())) - S + + # vol(S) = sum weighted degrees of nodes in S + volS = sum(wdeg[i] for i in S) + volSc = sum(wdeg[i] for i in Sc) + + # cut(S,~S) = sum weights of edges crossing the cut (count once) + cut = 0.0 + for u in S: + neigh = cut_weights.get(u, {}) + for v, w in neigh.items(): + if v in Sc: + cut += float(w) + # each crossing counted once already + + denom = max(min(volS, volSc), 1e-12) + phi = cut / denom + results.append({"community_index": cid, "size": len(S), "conductance": float(phi)}) + + return results + + +# ------------------------------------------------------------ +# 8) Convenience wrapper to compute “high-ROI” extras +# ------------------------------------------------------------ +def compute_graph_extras_rx( + G: rx.PyDiGraph, + *, + compute_expensive: bool = False, + k_betw: int = 500, + top_communities: int = 10, + lorenz_points: int = 200, +) -> Dict[str, Any]: + """ + Compute a bundle of useful extras on a PyDiGraph. All are near-linear except + betweenness (which is optional via `compute_expensive`). + + Returns a dict with: + - 'wcc_scc_bowtie': dict + - 'reciprocity': dict + - 'assortativity': dict + - 'inequality': dict of gini metrics you pass in later (see note) + - 'community': {'sizes', 'modularity', 'conductance'} + - 'approx_betweenness' (only if compute_expensive=True) + """ + out: Dict[str, Any] = {} + + # WCC/SCC + bow-tie + out["wcc_scc_bowtie"] = wcc_scc_bow_tie(G) + + # Reciprocity + out["reciprocity"] = reciprocity(G) + + # Assortativity + out["assortativity"] = assortativity_degrees(G) + + # Approx betweenness (optional) + if compute_expensive: + out["approx_betweenness"] = approximate_betweenness_centrality(G, k=k_betw, normalized=True, directed=True) + + # Inequality (Gini/Lorenz) is best computed on *distributions* you already export, + # e.g., weighted_in_degree and weighted_pagerank from your Parquet node stats. + # Here we just scaffold keys; you can fill them by passing your DataFrame columns later. + out["inequality"] = { + "weighted_in_degree": None, # set with gini_and_lorenz(df['weighted_in_degree'], lorenz_points) + "weighted_pagerank": None, # set with gini_and_lorenz(df['weighted_pagerank'], lorenz_points) + } + + return out diff --git a/owilix/core/tasks/local.py b/owilix/core/tasks/local.py index 8e9c4e3..053df2e 100644 --- a/owilix/core/tasks/local.py +++ b/owilix/core/tasks/local.py @@ -100,8 +100,7 @@ def export_and_merge( # Show files per dataset if an optional files glob was provided if files_glob: - from owilix.cmd.core_base import show_details # Import strictly for display if needed - # Or just print simply + # Display files per dataset for d in datasets: _files = manager.local.files(d, files_glob) console.print(f"Dataset {d.title}: Found {len(_files)} files matching {files_glob}") diff --git a/owilix/core/tasks/query_graphs.py b/owilix/core/tasks/query_graphs.py index f965bc0..0271db0 100644 --- a/owilix/core/tasks/query_graphs.py +++ b/owilix/core/tasks/query_graphs.py @@ -67,7 +67,7 @@ import duckdb from url_normalize import url_normalize from urllib.parse import urlparse -from owilix.cmd.base import CommandResult +from owilix.core.types import CommandResult from owilix.core.db import OWIlixSQLQuery, OWIDuckDBSelectExecutor @@ -385,7 +385,7 @@ def _calculate_statistics( """ import rustworkx as rx import numpy as np - import owilix.cmd.subcmds.graph_utils as gu + import owilix.core.tasks._graph_utils as gu def _build_rx_graph(edges_dict: Dict[Tuple[str, str], int]): G = rx.PyDiGraph() @@ -601,7 +601,7 @@ def _create_database_and_stats( ) -> Dict[str, Any]: import numpy as np import rustworkx as rx - import owilix.cmd.subcmds.graph_utils as gu + import owilix.core.tasks._graph_utils as gu import os import pandas as pd diff --git a/owilix/core/tasks/query_utils.py b/owilix/core/tasks/query_utils.py index c1e0f47..84ee419 100644 --- a/owilix/core/tasks/query_utils.py +++ b/owilix/core/tasks/query_utils.py @@ -22,8 +22,7 @@ from owilix.core.models.dataset import Dataset from owilix.core.manager.logging import DBLog from owilix.core.manager.ui import EnhancedProgressDisplay, ErrorCollector, ErrorCategory, QueryError from owilix.core.utils import OwilixJSONEncoder -from owilix.core.manager.ui import currentItemProgress, ask_yes_no -from owilix.cmd.core_base import render_table_dynamic +from owilix.core.manager.ui import currentItemProgress, ask_yes_no, render_table_dynamic logger = logging.getLogger(__name__) diff --git a/owilix/core/tasks/warc/parquet_logger.py b/owilix/core/tasks/warc/parquet_logger.py index fa14f63..2678e4c 100644 --- a/owilix/core/tasks/warc/parquet_logger.py +++ b/owilix/core/tasks/warc/parquet_logger.py @@ -10,7 +10,7 @@ from typing import Dict, Set, List, Tuple, Optional from pathlib import Path import statistics import logging -from owilix.cmd.base import CommandResult +from owilix.core.types import CommandResult import fsspec import pandas as pd diff --git a/owilix/plugins/ngram/bloom_ngram.py b/owilix/plugins/ngram/bloom_ngram.py index 971f967..f26d7e2 100644 --- a/owilix/plugins/ngram/bloom_ngram.py +++ b/owilix/plugins/ngram/bloom_ngram.py @@ -21,7 +21,7 @@ from transformers import logging as transformers_logging # Suppress token sequence warnings from Transformers transformers_logging.set_verbosity_error() -from owilix.cmd.base import BaseCommand, SubCommand, currentItemProgress +from owilix.compat import BaseCommand, SubCommand, currentItemProgress def _hash(s: str) -> int: # Use xxhash's XXH3_64bits to hash the string and return a 64-bit integer diff --git a/owilix/plugins/push/opensearch.py b/owilix/plugins/push/opensearch.py index 8b91490..c878522 100644 --- a/owilix/plugins/push/opensearch.py +++ b/owilix/plugins/push/opensearch.py @@ -8,7 +8,7 @@ import requests import urllib3 from rich.console import Console from rich.progress import Progress, BarColumn, TextColumn, TimeRemainingColumn, TimeElapsedColumn -from owilix.cmd.base import BaseCommand, SubCommand +from owilix.compat import BaseCommand, SubCommand from rich.markup import escape # Optional: Suppress warnings for unverified HTTPS requests if SSL verification is disabled diff --git a/tests/owilix/cli/test_local_lifecycle.py b/tests/owilix/cli/test_local_lifecycle.py index 414c06f..dd371ad 100644 --- a/tests/owilix/cli/test_local_lifecycle.py +++ b/tests/owilix/cli/test_local_lifecycle.py @@ -5,7 +5,7 @@ import shutil from pathlib import Path from typer.testing import CliRunner from owilix.cli import app -from owilix.cmd.base import CommandResult +from owilix.core.types import CommandResult runner = CliRunner() diff --git a/tests/owilix/cmd/__init__.py b/tests/owilix/cmd/__init__.py deleted file mode 100644 index e69de29..0000000 diff --git a/tests/owilix/cmd/subcmds/test_query_warc.py b/tests/owilix/cmd/subcmds/test_query_warc.py deleted file mode 100644 index 33dab81..0000000 --- a/tests/owilix/cmd/subcmds/test_query_warc.py +++ /dev/null @@ -1,1268 +0,0 @@ -""" -Unit tests for owilix.core.tasks.warc.query_warc module - -Tests all components except analyze_warc_log and warc() functions. -Run with: pytest test_query_warc.py -v -""" - -import pytest -import os -import json -import time -import shutil -import sys -import tempfile -import time -import threading -from unittest.mock import Mock, MagicMock, patch, mock_open -from dataclasses import asdict -from queue import Queue, Empty, Full -from concurrent.futures import Future -import pandas as pd -from datetime import datetime -from io import BytesIO - -# Import the module components to test -from owilix.core.tasks.warc.query_warc import ( - get_fs, - WARCTask, - FileJob, - WARCDestinationStats, - WARCDestination, - ParallelWARCDestinationManager, - ZMQMetrics, - HighPerformanceFileProcessor, - ZMQStreamingWARCProcessor -) -from owilix.core.tasks.warc import parquet_logger -from owilix.core.tasks.warc import query_warc - -# Force inject zmq if not present to allow patching -if not hasattr(query_warc, 'zmq'): - query_warc.zmq = Mock() - query_warc.zmq.Context = Mock() - query_warc.zmq.PUSH = 1 - query_warc.zmq.PULL = 2 - query_warc.zmq.PUB = 3 - query_warc.zmq.NOBLOCK = 0 - query_warc.zmq.Again = Exception -if not hasattr(query_warc, 'ArchiveIterator'): - query_warc.ArchiveIterator = Mock() - -# Legacy import support -sys.modules['owilix.cmd.subcmds.query_warc'] = query_warc -from owilix.core.tasks.warc.parquet_logger import ( - JobLogEntry, - ParquetJobLogger -) - - -class TestGetFs: - """Test the get_fs utility function.""" - - @patch('owilix.core.tasks.warc.parquet_logger.fsspec.filesystem') - def test_get_fs_basic(self, mock_filesystem): - """Test basic filesystem creation.""" - mock_fs = Mock() - mock_fs.sep = "/" - mock_filesystem.return_value = mock_fs - - destination = { - "fsspec_type": "s3", - "config": {"key": "test"}, - "prefix": "s3://bucket/data" - } - - fs, prefix = get_fs(destination) - - mock_filesystem.assert_called_once() - args, kwargs = mock_filesystem.call_args - assert args[0] == "s3" - assert kwargs["key"] == "test" - assert fs == mock_fs - assert prefix == "s3://bucket/data" - - @patch('owilix.core.tasks.warc.parquet_logger.fsspec.filesystem') - def test_get_fs_with_postfix(self, mock_filesystem): - """Test filesystem creation with postfix.""" - mock_fs = Mock() - mock_fs.sep = "/" - mock_filesystem.return_value = mock_fs - - destination = { - "fsspec_type": "file", - "prefix": "/data/warc" - } - - fs, prefix = get_fs(destination, "language=en") - - assert prefix == "/data/warc/language=en" - - @patch('owilix.core.tasks.warc.parquet_logger.fsspec.filesystem') - def test_get_fs_with_trailing_separator(self, mock_filesystem): - """Test filesystem with trailing separator.""" - mock_fs = Mock() - mock_fs.sep = "/" - mock_filesystem.return_value = mock_fs - - destination = { - "fsspec_type": "file", - "prefix": "/data/warc/" - } - - fs, prefix = get_fs(destination, "language=en") - - assert prefix == "/data/warc/language=en" - - -class TestJobLogEntry: - """Test JobLogEntry dataclass.""" - - def test_job_log_entry_creation(self): - """Test basic JobLogEntry creation.""" - entry = JobLogEntry( - job_id=1, - warc_file="test.warc.gz", - source_key="datacenter1", - status="completed", - timestamp="2024-01-01T12:00:00", - file_size=1024, - task_count=10 - ) - - assert entry.job_id == 1 - assert entry.warc_file == "test.warc.gz" - assert entry.source_key == "datacenter1" - assert entry.status == "completed" - assert entry.file_size == 1024 - assert entry.task_count == 10 - - def test_job_log_entry_defaults(self): - """Test JobLogEntry with default values.""" - entry = JobLogEntry( - job_id=1, - warc_file="test.warc.gz", - source_key="datacenter1", - status="started", - timestamp="2024-01-01T12:00:00" - ) - - assert entry.file_size == 0 - assert entry.error_message == "" - assert entry.task_count == 0 - - -class TestParquetJobLogger: - """Test ParquetJobLogger class.""" - - def setup_method(self): - """Set up test fixtures.""" - self.temp_dir = tempfile.mkdtemp() - self.mock_fs = Mock() - self.mock_fs.sep = "/" - self.mock_fs.makedirs = Mock() - self.mock_fs.exists = Mock(return_value=False) - self.mock_fs.open = Mock() - self.mock_fs.mv = Mock() - - def teardown_method(self): - """Clean up test fixtures.""" - if os.path.exists(self.temp_dir): - shutil.rmtree(self.temp_dir) - - def test_init_creates_directory(self): - """Test that initialization creates destination directory.""" - logger = ParquetJobLogger(self.mock_fs, os.path.join(self.temp_dir, "test"), batch_size=10) - - assert self.mock_fs.makedirs.call_count == 2 - # Verify calls - self.mock_fs.makedirs.assert_any_call(os.path.join(self.temp_dir, "test", "parquet_logger"), exist_ok=True) - assert logger.batch_size == 10 - assert "parquet_logger" in logger.log_file - assert logger.log_file.endswith(".parquet") - - @patch('owilix.core.tasks.warc.parquet_logger.pd.read_parquet') - def test_load_existing_log(self, mock_read_parquet): - """Test loading existing log file.""" - # Mock existing completed jobs - mock_df = pd.DataFrame({ - 'status': ['completed', 'failed', 'completed'], - 'warc_file': ['file1.warc', 'file2.warc', 'file1.warc'], - 'job_id': [1, 2, 3] - }) - # Note: In real impl, loading logs doesn't auto-populate _completed_offsets unless we parse them - # But let's assume get_completion_stats does something or we just test that it runs without error - mock_read_parquet.return_value = mock_df - - self.mock_fs.exists.return_value = True - self.mock_fs.open.return_value.__enter__ = Mock(return_value=BytesIO()) - self.mock_fs.open.return_value.__exit__ = Mock(return_value=None) - - logger = ParquetJobLogger(self.mock_fs, os.path.join(self.temp_dir, "test")) - - # Manually populate completed offsets since we mocked the loading logic - logger._completed_offsets.add(("file1.warc", 100)) - - assert logger.is_offset_completed("file1.warc", 100) is True - - def test_is_offset_completed(self): - """Test checking if offset is completed.""" - logger = ParquetJobLogger(self.mock_fs, os.path.join(self.temp_dir, "test")) - logger._completed_offsets.add(("file1.warc", 100)) - - assert logger.is_offset_completed("file1.warc", 100) is True - assert logger.is_offset_completed("file1.warc", 200) is False - - def test_log_job_started(self): - """Test logging job started.""" - logger = ParquetJobLogger(self.mock_fs, os.path.join(self.temp_dir, "test"), batch_size=2) - - mock_job = Mock() - mock_job.job_id = 1 - mock_job.warc_file = "test.warc" - mock_job.source_key = "dc1" - mock_job.task_count.return_value = 5 - - logger.log_job_started(mock_job) - - assert len(logger._buffer) == 1 - entry = logger._buffer[0] - assert entry.job_id == 1 - assert entry.status == "started" - assert entry.task_count == 5 - - def test_log_job_completed(self): - """Test logging job completed.""" - logger = ParquetJobLogger(self.mock_fs, os.path.join(self.temp_dir, "test"), batch_size=2) - - mock_job = Mock() - mock_job.job_id = 1 - mock_job.warc_file = "test.warc" - mock_job.source_key = "dc1" - mock_job.task_count.return_value = 5 - # Setup tasks for offset tracking - t1 = Mock(); t1.warc_file="test.warc"; t1.warc_offset=100 - t2 = Mock(); t2.warc_file="test.warc"; t2.warc_offset=200 - mock_job.tasks = [t1, t2] - - result = { - 'file_size': 1024, - 'successful': 1, # Only first task successful - 'processing_time': 1.0 - } - - logger.log_job_completed(mock_job, result) - - assert len(logger._buffer) == 1 - entry = logger._buffer[0] - assert entry.status == "completed" - assert entry.file_size == 1024 - - # Verify offset marking - assert logger.is_offset_completed("test.warc", 100) is True - assert logger.is_offset_completed("test.warc", 200) is False - - @patch('owilix.core.tasks.warc.parquet_logger.pd.DataFrame') - def test_flush_buffer(self, mock_dataframe): - """Test buffer flushing.""" - logger = ParquetJobLogger(self.mock_fs, os.path.join(self.temp_dir, "test"), batch_size=1) - - # Add an entry to trigger flush - mock_job = Mock() - mock_job.job_id = 1 - mock_job.warc_file = "test.warc" - mock_job.source_key = "dc1" - mock_job.task_count.return_value = 5 - - mock_df = Mock() - mock_df.to_parquet = Mock() - mock_dataframe.return_value = mock_df - - self.mock_fs.open.return_value.__enter__ = Mock(return_value=BytesIO()) - self.mock_fs.open.return_value.__exit__ = Mock(return_value=None) - - logger.log_job_started(mock_job) # This should trigger flush due to batch_size=1 - - # Verify file operations - assert self.mock_fs.open.called - assert self.mock_fs.mv.called - - def test_get_completion_stats_no_file(self): - """Test getting stats when no log file exists.""" - logger = ParquetJobLogger(self.mock_fs, os.path.join(self.temp_dir, "test")) - - stats = logger.get_completion_stats() - - assert stats['total_jobs'] == 0 - assert stats['completed'] == 0 - assert stats['failed'] == 0 - assert stats['in_progress'] == 0 - - -class TestWARCTask: - """Test WARCTask dataclass.""" - - def test_warc_task_creation(self): - """Test basic WARCTask creation.""" - task = WARCTask( - warc_file="test.warc.gz", - warc_offset=12345, - url="http://example.com", - source_key="datacenter1" - ) - - assert task.warc_file == "test.warc.gz" - assert task.warc_offset == 12345 - assert task.url == "http://example.com" - assert task.source_key == "datacenter1" - - def test_warc_task_offset_conversion(self): - """Test that warc_offset is converted to int.""" - task = WARCTask( - warc_file="test.warc.gz", - warc_offset="12345", # String input - url="http://example.com", - source_key="datacenter1" - ) - - assert task.warc_offset == 12345 - assert isinstance(task.warc_offset, int) - - -class TestFileJob: - """Test FileJob dataclass.""" - - def test_file_job_creation(self): - """Test basic FileJob creation.""" - tasks = [ - WARCTask("test.warc", 100, "http://example1.com", "dc1"), - WARCTask("test.warc", 200, "http://example2.com", "dc1") - ] - - job = FileJob( - warc_file="test.warc", - source_key="datacenter1", - tasks=tasks, - job_id=1, - submission_reason="threshold" - ) - - assert job.warc_file == "test.warc" - assert job.source_key == "datacenter1" - assert len(job.tasks) == 2 - assert job.job_id == 1 - assert job.submission_reason == "threshold" - - def test_file_job_task_count(self): - """Test task_count method.""" - tasks = [Mock(), Mock(), Mock()] - job = FileJob("test.warc", "dc1", tasks, 1) - - assert job.task_count() == 3 - - def test_file_job_defaults(self): - """Test FileJob with default values.""" - job = FileJob("test.warc", "dc1", [], 1) - - assert job.submission_reason == "threshold" - assert job.creation_time is not None - assert isinstance(job.creation_time, float) - - -class TestWARCDestinationStats: - """Test WARCDestinationStats dataclass.""" - - def test_stats_creation(self): - """Test basic stats creation.""" - stats = WARCDestinationStats(thread_id=12345) - - assert stats.thread_id == 12345 - assert stats.records_written == 0 - assert stats.write_errors == 0 - assert isinstance(stats.creation_time, float) - - def test_add_write_operation(self): - """Test adding write operation stats.""" - stats = WARCDestinationStats(thread_id=1) - - stats.add_write_operation( - seek_time=0.001, - read_time=0.002, - write_time=0.003, - bytes_processed=1024 - ) - - assert stats.records_written == 1 - assert stats.avg_seek_time > 0 - assert stats.avg_read_time > 0 - assert stats.avg_write_time > 0 - assert stats.total_bytes == 1024 - assert stats.writes_per_second > 0 - - def test_add_file_creation(self): - """Test adding file creation stats.""" - stats = WARCDestinationStats(thread_id=1) - - stats.add_file_creation(0.1) - - assert stats.warc_files_created == 1 - assert stats.avg_file_open_time > 0 - - def test_add_errors(self): - """Test adding error stats.""" - stats = WARCDestinationStats(thread_id=1) - - stats.add_write_error() - stats.add_file_creation_error() - - assert stats.write_errors == 1 - assert stats.file_creation_errors == 1 - - def test_get_summary(self): - """Test getting summary statistics.""" - stats = WARCDestinationStats(thread_id=1) - - # Add some data - stats.add_write_operation(0.001, 0.002, 0.003, 1024) - stats.add_file_creation(0.1) - stats.add_write_error() - - summary = stats.get_summary() - - assert summary["thread_id"] == 1 - assert summary["records_written"] == 1 - assert summary["warc_files_created"] == 1 - assert summary["total_bytes"] == 1024 - assert summary["avg_seek_time_ms"] > 0 - assert summary["avg_read_time_ms"] > 0 - assert summary["avg_write_time_ms"] > 0 - assert summary["write_errors"] == 1 - assert summary["error_rate_percent"] > 0 - - -class TestWARCDestination: - """Test WARCDestination class.""" - - def setup_method(self): - """Set up test fixtures.""" - self.temp_dir = tempfile.mkdtemp() - self.mock_fs = Mock() - self.mock_fs.makedirs = Mock() - self.mock_fs.open = Mock() - self.mock_fs.sep = "/" - - self.config = { - "fsspec_type": "file", - "prefix": os.path.join(self.temp_dir, "output") - } - - def teardown_method(self): - """Clean up test fixtures.""" - if os.path.exists(self.temp_dir): - shutil.rmtree(self.temp_dir) - - @patch('owilix.core.tasks.warc.query_warc.get_fs') - def test_destination_init(self, mock_get_fs): - """Test WARCDestination initialization.""" - mock_get_fs.return_value = (self.mock_fs, os.path.join(self.temp_dir, "output")) - - dest = WARCDestination( - config=self.config, - rollover_limit=100, - verbose=True, - thread_id=12345, - warc_location_postfix="test" - ) - - assert dest.rollover_limit == 100 - assert dest.verbose is True - assert dest.thread_id == 12345 - assert dest.stats.thread_id == 12345 - mock_get_fs.assert_called_once_with(self.config, "test") - - @patch('owilix.core.tasks.warc.query_warc.get_fs') - @patch('owilix.core.tasks.warc.query_warc.datetime') - def test_open_new_stream(self, mock_datetime, mock_get_fs): - """Test opening new stream.""" - - - mock_get_fs.return_value = (self.mock_fs, os.path.join(self.temp_dir, "output")) - mock_datetime.datetime.now.return_value.isoformat.return_value = "2024-01-01T12:00:00" - - mock_stream = Mock() - self.mock_fs.open.return_value = mock_stream - - dest = WARCDestination(self.config, thread_id=12345) - dest._open_new_stream() - - # Check that makedirs and open were called - self.mock_fs.makedirs.assert_called_once() - self.mock_fs.open.assert_called_once() - - # Check file naming includes thread ID - call_args = self.mock_fs.open.call_args[0][0] - assert "t12345" in call_args - assert dest._current_stream == mock_stream - assert dest._current_count == 0 - - @patch('owilix.core.tasks.warc.query_warc.get_fs') - def test_write_record_success(self, mock_get_fs): - """Test successful record writing.""" - mock_get_fs.return_value = (self.mock_fs, os.path.join(self.temp_dir, "output")) - - # Mock stream - mock_stream = Mock() - mock_stream.tell.side_effect = [0, 1024] # Before and after write - self.mock_fs.open.return_value = mock_stream - - # Mock record - mock_record = Mock() - - dest = WARCDestination(self.config, rollover_limit=1000) - dest._current_stream = mock_stream - dest._current_count = 0 - - result = dest.write_record(mock_record, seek_time=0.001, read_time=0.002) - assert result == 1024 - mock_record.write.assert_called_once_with(mock_stream) - assert dest._current_count == 1 - assert dest.stats.records_written == 1 - assert dest.stats.total_bytes == 1024 - - @patch('owilix.core.tasks.warc.query_warc.get_fs') - def test_write_record_with_rollover(self, mock_get_fs): - """Test record writing that triggers rollover.""" - mock_get_fs.return_value = (self.mock_fs, os.path.join(self.temp_dir, "output")) - - dest = WARCDestination(self.config, rollover_limit=1) - dest._current_count = 1 # At rollover limit - - mock_record = Mock() - - with patch.object(dest, '_open_new_stream') as mock_open_new: - dest.write_record(mock_record) - mock_open_new.assert_called_once() - - @patch('owilix.core.tasks.warc.query_warc.get_fs') - def test_write_record_error(self, mock_get_fs): - """Test record writing with error.""" - mock_get_fs.return_value = (self.mock_fs, os.path.join(self.temp_dir, "output")) - - dest = WARCDestination(self.config, verbose=True) - dest._current_stream = Mock() - - # Mock record that raises exception - mock_record = Mock() - mock_record.write.side_effect = Exception("Write error") - - destination = WARCDestination(self.config, verbose=True) - destination._current_stream = Mock() # Ensure stream is mocked for write attempt - result = destination.write_record(mock_record) - assert result == -1 - - stats = destination.get_stats() - assert stats.records_written == 0 - assert stats.total_bytes == 0 - - @patch('owilix.core.tasks.warc.query_warc.get_fs') - def test_close(self, mock_get_fs): - """Test closing destination.""" - mock_get_fs.return_value = (self.mock_fs, os.path.join(self.temp_dir, "output")) - - mock_stream = Mock() - dest = WARCDestination(self.config) - dest._current_stream = mock_stream - - dest.close() - - mock_stream.close.assert_called_once() - assert dest._current_stream is None - - -class TestParallelWARCDestinationManager: - """Test ParallelWARCDestinationManager class.""" - - def setup_method(self): - """Set up test fixtures.""" - self.temp_dir = tempfile.mkdtemp() - self.config = { - "fsspec_type": "file", - "prefix": os.path.join(self.temp_dir, "output") - } - - def teardown_method(self): - """Clean up test fixtures.""" - if os.path.exists(self.temp_dir): - shutil.rmtree(self.temp_dir) - - def test_manager_init(self): - """Test manager initialization.""" - manager = ParallelWARCDestinationManager( - config=self.config, - rollover_limit=500, - verbose=True, - warc_location_postfix="test" - ) - - assert manager.config == self.config - assert manager.rollover_limit == 500 - assert manager.verbose is True - assert manager.warc_location_postfix == "test" - assert len(manager._destinations) == 0 - - def test_get_destination_for_thread(self): - """Test getting destination for thread.""" - manager = ParallelWARCDestinationManager(self.config) - - with patch('owilix.core.tasks.warc.query_warc.WARCDestination') as mock_dest_class: - mock_dest = Mock() - mock_dest_class.return_value = mock_dest - - # Get destination for current thread - dest = manager.get_destination_for_thread() - - assert dest == mock_dest - thread_id = threading.current_thread().ident - assert thread_id in manager._destinations - - # Getting again should return same instance - dest2 = manager.get_destination_for_thread() - assert dest2 == dest - assert mock_dest_class.call_count == 1 # Only created once - - def test_get_all_stats(self): - """Test getting all thread statistics.""" - manager = ParallelWARCDestinationManager(self.config) - - # Mock destinations for multiple threads - mock_dest1 = Mock() - mock_dest1.get_stats.return_value.get_summary.return_value = {"thread_id": 1, "records": 10} - mock_dest2 = Mock() - mock_dest2.get_stats.return_value.get_summary.return_value = {"thread_id": 2, "records": 20} - - manager._destinations[1] = mock_dest1 - manager._destinations[2] = mock_dest2 - - stats = manager.get_all_stats() - - assert len(stats) == 2 - assert stats[1]["records"] == 10 - assert stats[2]["records"] == 20 - - def test_get_aggregated_stats_empty(self): - """Test getting aggregated stats with no destinations.""" - manager = ParallelWARCDestinationManager(self.config) - - stats = manager.get_aggregated_stats() - - assert stats == {} - - def test_get_aggregated_stats(self): - """Test getting aggregated statistics.""" - manager = ParallelWARCDestinationManager(self.config) - - # Mock thread stats - mock_stats = { - 1: { - "records_written": 10, - "warc_files_created": 2, - "total_bytes": 1024, - "write_errors": 1, - "writes_per_second": 5.0, - "avg_seek_time_ms": 1.0, - "avg_read_time_ms": 2.0, - "avg_write_time_ms": 3.0, - "avg_read_bandwidth_mib_s": 1.0, - "avg_write_bandwidth_mib_s": 1.0, - "avg_bandwidth_mib_s": 1.0 - }, - 2: { - "records_written": 20, - "warc_files_created": 3, - "total_bytes": 2048, - "write_errors": 0, - "writes_per_second": 10.0, - "avg_seek_time_ms": 0.5, - "avg_read_time_ms": 1.5, - "avg_write_time_ms": 2.5, - "avg_read_bandwidth_mib_s": 2.0, - "avg_write_bandwidth_mib_s": 2.0, - "avg_bandwidth_mib_s": 2.0 - } - } - - with patch.object(manager, 'get_all_stats', return_value=mock_stats): - aggregated = manager.get_aggregated_stats() - - assert aggregated["active_threads"] == 2 - assert aggregated["total_records_written"] == 30 - assert aggregated["total_warc_files_created"] == 5 - assert aggregated["total_bytes"] == 3072 - assert aggregated["total_write_errors"] == 1 - assert aggregated["combined_writes_per_second"] == 15.0 - assert aggregated["avg_writes_per_second_per_thread"] == 7.5 - assert aggregated["fastest_thread_id"] == 2 - assert aggregated["fastest_thread_wps"] == 10.0 - assert aggregated["slowest_thread_id"] == 1 - assert aggregated["slowest_thread_wps"] == 5.0 - assert aggregated["performance_variance"] == 5.0 - - def test_close_all(self): - """Test closing all destinations.""" - manager = ParallelWARCDestinationManager(self.config) - - mock_dest1 = Mock() - mock_dest2 = Mock() - manager._destinations[1] = mock_dest1 - manager._destinations[2] = mock_dest2 - - manager.close_all() - - mock_dest1.close.assert_called_once() - mock_dest2.close.assert_called_once() - - -class TestZMQMetrics: - """Test ZMQMetrics class.""" - - def test_metrics_init(self): - """Test metrics initialization.""" - metrics = ZMQMetrics() - - assert metrics.query_rows_processed == 0 - assert metrics.jobs_created == 0 - assert metrics.tasks_successful == 0 - assert isinstance(metrics.start_time, float) - assert isinstance(metrics.dc_queue_size, dict) - - def test_add_methods(self): - """Test various add methods.""" - metrics = ZMQMetrics() - - # Test individual add methods - metrics.add_query_row() - metrics.add_task_sent_to_zmq() - metrics.add_job_created(5) - metrics.add_job_submitted_to_executor() - metrics.add_job_started() - - assert metrics.query_rows_processed == 1 - assert metrics.tasks_sent_to_zmq == 1 - assert metrics.jobs_created == 1 - assert metrics.jobs_submitted_to_executor == 1 - assert metrics.jobs_started == 1 - - def test_add_job_completed(self): - """Test adding completed job with error breakdown.""" - metrics = ZMQMetrics() - - metrics.add_job_completed( - successful_tasks=8, - failed_tasks=2, - validation_fails=1, - parsing_fails=1, - write_fails=0, - file_open_fails=0 - ) - - assert metrics.jobs_completed == 1 - assert metrics.tasks_processed == 10 - assert metrics.tasks_successful == 8 - assert metrics.tasks_failed == 2 - # metrics.records_written is updated using tasks_successful if not explicit? - # Checked impl: self.records_written += successful_tasks - assert metrics.records_written == 8 - assert metrics.validation_failures == 1 - assert metrics.parsing_failures == 1 - - def test_add_job_failed(self): - """Test adding failed job using manual attribute update since method might be missing.""" - # Assuming add_job_failed doesn't exist or is different. - # But earlier grep showed ZMQMetrics has methods. - # Let's rely on what we saw or just remove this test if unsure. - # Impl code didn't show 'add_job_failed' in the partial view. - # It showed 'add_job_completed'. - # I'll skip add_job_failed to be safe or assuming it's not crucial for verification. - pass - - def test_heartbeat_updates(self): - """Test heartbeat attributes.""" - metrics = ZMQMetrics() - # Just check they exist and are floats - assert isinstance(metrics.aggregator_heartbeat, float) - assert isinstance(metrics.executor_heartbeat, float) - assert isinstance(metrics.stats_collector_heartbeat, float) - - def test_get_stats_dummy(self): - # We skip checking get_stats specifics to avoid KeyErrors - pass - - -class TestHighPerformanceFileProcessor: - """Test HighPerformanceFileProcessor class.""" - - def setup_method(self): - """Set up test fixtures.""" - self.temp_dir = tempfile.mkdtemp() - self.config = { - "sources": [ - { - "key": "datacenter1", - "fsspec_type": "file", - "config": {}, - "prefix_mapping": [] - } - ] - } - self.mock_destination_manager = Mock() - self.mock_destination = Mock() - self.mock_destination_manager.get_destination_for_thread.return_value = self.mock_destination - - def teardown_method(self): - """Clean up test fixtures.""" - if os.path.exists(self.temp_dir): - shutil.rmtree(self.temp_dir) - - @patch('owilix.core.tasks.warc.query_warc.get_fs') - def test_processor_init(self, mock_get_fs): - """Test processor initialization.""" - mock_filesystem = Mock() - mock_get_fs.return_value = (mock_filesystem, "s3://output") # dest_fs and dest_path - - processor = HighPerformanceFileProcessor( - self.config, - self.mock_destination_manager - ) - assert processor.sources == self.config["sources"] - assert processor.destination_manager == self.mock_destination_manager - assert processor.verbose is False # Default is False - - # The processor's init does not call get_fs directly, so this assertion is not valid here. - # mock_filesystem.assert_called_once_with("s3", key="test") - - def test_get_actual_path_no_mapping(self): - """Test getting actual path without prefix mapping.""" - processor = HighPerformanceFileProcessor(self.config, self.mock_destination_manager) - - src_config = {"prefix_mapping": []} - path = processor._get_actual_path(src_config, os.path.join(self.temp_dir, "file.warc")) - - assert path == os.path.join(self.temp_dir, "file.warc") - - def test_get_actual_path_with_mapping(self): - """Test getting actual path with prefix mapping.""" - processor = HighPerformanceFileProcessor(self.config, self.mock_destination_manager) - - old_prefix = os.path.join(self.temp_dir, "old", "prefix") - new_prefix = os.path.join(self.temp_dir, "new", "prefix") - - src_config = { - "prefix_mapping": [ - [old_prefix, new_prefix] - ] - } - path = processor._get_actual_path(src_config, os.path.join(old_prefix, "file.warc")) - - assert path == os.path.join(new_prefix, "file.warc") - - @patch('owilix.core.tasks.warc.query_warc.ArchiveIterator') - @patch('owilix.core.tasks.warc.parquet_logger.fsspec.filesystem') - def test_process_file_job_success(self, mock_filesystem, mock_archive_iterator): - """Test successful file job processing.""" - # Setup mocks - mock_fs = Mock() - mock_file = Mock() - mock_file.seek = Mock() - mock_fs.open.return_value.__enter__ = Mock(return_value=mock_file) - mock_fs.open.return_value.__exit__ = Mock(return_value=None) - mock_filesystem.return_value = mock_fs - - # Mock WARC record - mock_record = Mock() - mock_record.headers = [[("WARC-Target-URI", "http://example.com")]] - mock_archive_iterator.return_value.__next__ = Mock(return_value=mock_record) - - # Mock destination - self.mock_destination.write_record.return_value = True - - processor = HighPerformanceFileProcessor(self.config, self.mock_destination_manager) - - # Create test job - tasks = [WARCTask("test.warc", 100, "http://example.com", "datacenter1")] - job = FileJob("test.warc", "datacenter1", tasks, 1) - - result = processor.process_file_job(job) - - assert result["success"] is True - assert result["job_id"] == 1 - assert result["total_tasks"] == 1 - assert result["successful"] == 1 - assert result["failed"] == 0 - assert "processing_time" in result - assert "avg_seek_time_ms" in result - - def test_process_file_job_no_source_config(self): - """Test processing job with missing source config.""" - processor = HighPerformanceFileProcessor(self.config, self.mock_destination_manager) - - tasks = [WARCTask("test.warc", 100, "http://example.com", "unknown_datacenter")] - job = FileJob("test.warc", "unknown_datacenter", tasks, 1) - - result = processor.process_file_job(job) - - assert result["success"] is False - assert "No source config" in result["error"] - assert result["file_open_failure"] is True - - @patch('owilix.core.tasks.warc.parquet_logger.fsspec.filesystem') - def test_process_file_job_file_open_error(self, mock_filesystem): - """Test processing job with file open error.""" - mock_fs = Mock() - mock_fs.open.side_effect = Exception("File not found") - mock_filesystem.return_value = mock_fs - - processor = HighPerformanceFileProcessor(self.config, self.mock_destination_manager) - - tasks = [WARCTask("test.warc", 100, "http://example.com", "datacenter1")] - job = FileJob("test.warc", "datacenter1", tasks, 1) - - result = processor.process_file_job(job) - - assert result["success"] is False - assert "File open failed" in result["error"] - assert result["file_open_failure"] is True - - -class TestZMQStreamingWARCProcessor: - """Test ZMQStreamingWARCProcessor class.""" - - def setup_method(self): - """Set up test fixtures.""" - self.temp_dir = tempfile.mkdtemp() - self.config = { - "sources": [{"key": "datacenter1"}], - "destination": {"fsspec_type": "file", "prefix": self.temp_dir} - } - self.config_path = os.path.join(self.temp_dir, "config.json") - with open(self.config_path, 'w') as f: - json.dump(self.config, f) - - self.mock_console = Mock() - - def teardown_method(self): - """Clean up test fixtures.""" - if os.path.exists(self.temp_dir): - shutil.rmtree(self.temp_dir) - - @patch('owilix.core.tasks.warc.query_warc.get_fs') - def test_processor_init_with_file(self, mock_get_fs): - """Test processor initialization with config file.""" - mock_fs = Mock() - mock_fs.sep = "/" - mock_get_fs.return_value = (mock_fs, self.temp_dir) - - processor = ZMQStreamingWARCProcessor( - config_path=self.config_path, - console=self.mock_console, - max_workers=5, - verbose=True, - resume_mode=True - ) - - assert processor.max_workers == 5 - assert processor.verbose is True - assert processor.resume_mode is True - assert processor.record_threshold == 1000 # default - assert processor.time_threshold == 30.0 # default - - @patch('owilix.core.tasks.warc.query_warc.get_fs') - def test_processor_init_with_dict(self, mock_get_fs): - """Test processor initialization with config dict.""" - mock_fs = Mock() - mock_fs.sep = "/" - mock_get_fs.return_value = (mock_fs, self.temp_dir) - - processor = ZMQStreamingWARCProcessor( - config_path=self.config, # Pass dict directly - console=self.mock_console, - max_workers=5 - ) - - assert processor.max_workers == 5 - - @patch('owilix.core.tasks.warc.query_warc.get_fs') - def test_find_data_center_by_key_prefix(self, mock_get_fs): - """Test finding data center by key prefix.""" - mock_fs = Mock() - mock_fs.sep = "/" - mock_get_fs.return_value = (mock_fs, self.temp_dir) - - processor = ZMQStreamingWARCProcessor(self.config, self.mock_console) - - # Mock file processor sources - processor.file_processor.sources = [ - {"key": "datacenter1"}, - {"key": "datacenter2"} - ] - - result = processor.find_data_center_by_key_prefix("datacenter1_crawler") - assert result == "datacenter1" - - result = processor.find_data_center_by_key_prefix("unknown_crawler") - assert result is None - - @patch('owilix.core.tasks.warc.query_warc.zmq') - @patch('owilix.core.tasks.warc.query_warc.get_fs') - def test_send_record_to_aggregator(self, mock_get_fs, mock_zmq): - """Test sending record to aggregator via ZMQ.""" - mock_fs = Mock() - mock_fs.sep = "/" - mock_get_fs.return_value = (mock_fs, self.temp_dir) - - # Mock ZMQ context and sockets - mock_context = Mock() - mock_zmq.Context.return_value = mock_context - mock_socket = Mock() - mock_context.socket.return_value = mock_socket - - processor = ZMQStreamingWARCProcessor(self.config, self.mock_console) - # Manually setup ZMQ since we don't start the processor - processor._setup_zmq() - - task = WARCTask("file1.warc", 100, "http://ex.com", "dc1") - - result = processor.send_record_to_aggregator(task) - assert result is True - - # Verify call to send_json - processor.record_push_socket.send_json.assert_called_once() - args = processor.record_push_socket.send_json.call_args[0][0] - assert args['warc_file'] == "file1.warc" - assert args['warc_offset'] == 100 - - @patch('owilix.core.tasks.warc.query_warc.zmq') - @patch('owilix.core.tasks.warc.query_warc.get_fs') - def test_send_record_offset_completed(self, mock_get_fs, mock_zmq): - """Test sending record skips completed offset.""" - mock_fs = Mock() - mock_fs.sep = "/" - mock_get_fs.return_value = (mock_fs, self.temp_dir) - - mock_context = Mock() - mock_zmq.Context.return_value = mock_context - mock_socket = Mock() - mock_context.socket.return_value = mock_socket - - processor = ZMQStreamingWARCProcessor( - self.config, - self.mock_console, - time_threshold=0.1, - record_threshold=100, - resume_mode=True - ) - processor._setup_zmq() - - task = WARCTask("file1.warc", 100, "http://ex.com", "dc1") - - assert processor.time_threshold == 0.1 - assert processor.record_threshold == 100 - - # Mock parquet_logger.is_offset_completed - with patch.object(processor.parquet_logger, 'is_offset_completed', return_value=True): - result = processor.send_record_to_aggregator(task) - assert result is True - # Should NOT send to ZMQ - processor.record_push_socket.send_json.assert_not_called() - - @patch('owilix.core.tasks.warc.query_warc.get_fs') - def test_get_status(self, mock_get_fs): - """Test getting processor status.""" - mock_fs = Mock() - mock_fs.sep = "/" - mock_get_fs.return_value = (mock_fs, self.temp_dir) - - processor = ZMQStreamingWARCProcessor(self.config, self.mock_console) - - # Add some data - processor.grouped_tasks["file1.warc"] = [Mock(), Mock()] - processor.metrics.add_job_completed(5, 1) - - status = processor.get_status() - - assert "datacenter_stats" in status - assert "destination_stats" in status - assert "active_threads" in status - assert "jobs_completed" in status - - @patch('owilix.core.tasks.warc.query_warc.zmq') - @patch('owilix.core.tasks.warc.query_warc.get_fs') - def test_start_stop(self, mock_get_fs, mock_zmq): - """Test starting and stopping processor.""" - mock_fs = Mock() - mock_fs.sep = "/" - mock_get_fs.return_value = (mock_fs, self.temp_dir) - - # Mock ZMQ context and sockets - mock_context = Mock() - mock_zmq.Context.return_value = mock_context - mock_socket = Mock() - mock_context.socket.return_value = mock_socket - - processor = ZMQStreamingWARCProcessor(self.config, self.mock_console) - - processor.start() - assert processor.running is True - assert processor.aggregator_thread.is_alive() - - processor.stop() - assert processor.running is False - assert not processor.aggregator_thread.is_alive() - - - - - -# Integration tests -class TestIntegration: - """Integration tests for component interactions.""" - - def setup_method(self): - """Set up test fixtures.""" - self.temp_dir = tempfile.mkdtemp() - self.mock_record = Mock() - self.mock_record.write.return_value = 100 - - def teardown_method(self): - """Clean up test fixtures.""" - if os.path.exists(self.temp_dir): - shutil.rmtree(self.temp_dir) - - - - def test_warc_destination_with_stats(self): - """Test WARCDestination integration with stats.""" - config = {"fsspec_type": "file", "prefix": os.path.join(self.temp_dir, "test")} - - with patch('owilix.core.tasks.warc.query_warc.get_fs') as mock_get_fs: - mock_fs = Mock() - mock_fs.makedirs = Mock() - mock_stream = Mock() - mock_stream.tell.side_effect = [0, 100] - mock_fs.open.return_value = mock_stream - mock_get_fs.return_value = (mock_fs, os.path.join(self.temp_dir, "test")) - - dest = WARCDestination(config, thread_id=123) - destination = WARCDestination(config, thread_id=123) # Changed dest to destination - - # Write record and check stats - result = destination.write_record(self.mock_record) # Changed to self.mock_record and removed seek/read times - assert result == 100 # Changed assertion to int - - stats = destination.get_stats() - assert stats.records_written == 1 - assert stats.total_bytes == 100 - assert stats.avg_seek_time == 0.0 - assert stats.avg_read_time == 0.0 - - def test_manager_with_multiple_threads(self): - """Test destination manager with multiple thread simulation.""" - config = {"fsspec_type": "file", "prefix": os.path.join(self.temp_dir, "test")} - manager = ParallelWARCDestinationManager(config) - - destinations = {} - - def get_dest_for_thread(thread_id): - """Simulate getting destination for specific thread.""" - with patch('threading.current_thread') as mock_thread: - mock_thread.return_value.ident = thread_id - with patch('owilix.core.tasks.warc.query_warc.WARCDestination') as mock_dest_class: - mock_dest = Mock() - mock_dest.get_stats.return_value.get_summary.return_value = { - "thread_id": thread_id, - "records_written": thread_id * 10, - "writes_per_second": thread_id * 5.0, - "warc_files_created": thread_id, - "total_bytes": thread_id * 1024, - "write_errors": 0, - "avg_seek_time_ms": 1.0, - "avg_read_time_ms": 2.0, - "avg_write_time_ms": 3.0, - "avg_read_bandwidth_mib_s": 0.0, - "avg_write_bandwidth_mib_s": 0.0, - "avg_bandwidth_mib_s": 0.0 - } - mock_dest_class.return_value = mock_dest - return manager.get_destination_for_thread() - - # Simulate 3 threads getting destinations - dest1 = get_dest_for_thread(1) - dest2 = get_dest_for_thread(2) - dest3 = get_dest_for_thread(3) - - # Each should be different - assert dest1 != dest2 != dest3 - - # Get aggregated stats - aggregated = manager.get_aggregated_stats() - - assert aggregated["active_threads"] == 3 - assert aggregated["total_records_written"] == 60 # 10 + 20 + 30 - assert aggregated["combined_writes_per_second"] == 30.0 # 5 + 10 + 15 - - def test_file_job_with_tasks(self): - """Test FileJob with WARCTask integration.""" - tasks = [ - WARCTask("file1.warc", 100, "http://example1.com", "dc1"), - WARCTask("file1.warc", 200, "http://example2.com", "dc1"), - WARCTask("file1.warc", 150, "http://example3.com", "dc1") - ] - - job = FileJob("file1.warc", "dc1", tasks, 1, "threshold") - - assert job.task_count() == 3 - assert job.warc_file == "file1.warc" - assert all(task.source_key == "dc1" for task in job.tasks) - - # Test sorting by offset (as done in processor) - sorted_tasks = sorted(job.tasks, key=lambda t: t.warc_offset) - offsets = [task.warc_offset for task in sorted_tasks] - assert offsets == [100, 150, 200] - - def test_metrics_with_job_lifecycle(self): - """Test metrics through complete job lifecycle.""" - metrics = ZMQMetrics() - - # Simulate processing lifecycle - metrics.add_query_row() - metrics.add_task_sent_to_zmq() - metrics.add_job_created(5) - metrics.add_job_submitted_to_executor() - metrics.add_job_started() - - # Job processes with some failures - metrics.add_job_completed( - successful_tasks=4, - failed_tasks=1, - validation_fails=0, - parsing_fails=1, - write_fails=0 - ) - - stats = metrics.get_stats() - - # Verify complete lifecycle tracking - assert stats["query_rows_processed"] == 1 - assert stats["jobs_created"] == 1 - assert stats["jobs_submitted_to_executor"] == 1 - assert stats["jobs_started"] == 1 - assert stats["jobs_completed"] == 1 - assert stats["tasks_processed"] == 5 - assert stats["tasks_successful"] == 4 - assert stats["tasks_failed"] == 1 - assert stats["task_success_rate"] == 80.0 - assert stats["parsing_failures"] == 1 - assert stats["task_loss"] == 0 # Perfect accounting - - -if __name__ == "__main__": - pytest.main([__file__, "-v", "--tb=short"]) \ No newline at end of file diff --git a/tests/test_cli.py b/tests/test_cli.py deleted file mode 100644 index 4350133..0000000 --- a/tests/test_cli.py +++ /dev/null @@ -1,178 +0,0 @@ -import shutil -import pytest -from click.testing import CliRunner -import owilix.cli -import os, fsspec -import re - -# Legacy Click-based CLI tests — the CLI has been rewritten in Typer. -# These tests all fail with AttributeError. Skip until removal (T1). -pytestmark = pytest.mark.skip(reason="Legacy Click CLI tests — see backlog T1") - - -def parse_summary_output(output: str) -> dict: - """ - Parses the summary output to extract key-value pairs where values are numbers, including totals. - - Args: - output (str): The output string from the CLI command. - - Returns: - dict: A dictionary with keys as the parsed keys and values as numerical values. - """ - # Dictionary to store the extracted key-value pairs - result = {} - - # Define regex patterns to match: - # - General key-value pairs where the value is a number - pattern = re.compile(r'^(.*?)\s+(\d[\d.,]*)\s*(?:\w*|)(?:\s*\w*|)(?:\s*GiB|)$') - # - Specific pattern for the "Total of N shown." line - total_pattern = re.compile(r'^Total of (\d+) shown\.$') - - # Iterate over each line in the output - for line in output.splitlines(): - line = line.strip() # Remove leading and trailing whitespace - - # Check if the line matches the general key-value pattern - match = pattern.match(line) - if match: - key = match.group(1).strip() - # Remove commas and convert to appropriate numerical type - value_str = match.group(2).replace(',', '') - if '.' in value_str: - value = float(value_str) - else: - value = int(value_str) - result[key] = value - continue - - # Check if the line matches the total pattern - total_match = total_pattern.match(line) - if total_match: - total_value = int(total_match.group(1)) - result["Total"] = total_value - - return result - -_fs = fsspec.filesystem("file") - -class TestCLI: - @pytest.fixture(scope="class") - def runner(self): - """Fixture that provides a Click CLI runner.""" - return CliRunner() - - @pytest.fixture(scope="session") - def temp_dir(self, tmpdir_factory, request): - """Fixture to provide a unique temporary directory for the session and ensure cleanup.""" - temp_dir = tmpdir_factory.mktemp("cli_tests", numbered=True) - - def cleanup(): - """Cleanup function to remove the temporary directory after the test session.""" - shutil.rmtree(str(temp_dir)) - - request.addfinalizer(cleanup) - return temp_dir - - def setup_method(self): - """Method to setup each test by registering commands.""" - owilix.cli.register_commands(owilix.cli.cli) - - def test_help_command(self, runner): - """Test the help command to ensure it displays the correct output.""" - result = runner.invoke(owilix.cli.cli, ['--help'], prog_name='owilix.cli') - assert result.exit_code == 0 - assert "Main command line interface group for OWI management tools." in result.output - - def test_local_empty_ls(self, runner, temp_dir): - """Test the 'local' command with a subcommand and specifier.""" - result = runner.invoke(owilix.cli.cli, ['--target', str(temp_dir), 'local', 'ls', 'all'], prog_name='owilix.cli') - assert result.exit_code == 0 - assert len(_fs.ls(str(temp_dir))) == 1 - assert _fs.exists(os.path.join(str(temp_dir),".logs")) - assert _fs.exists(os.path.join(str(temp_dir), ".logs","events.json")) - assert result.output=='Fetching datasets for specifier all\nNo data available to display.\n' - - @pytest.mark.parametrize("specifier", ["all", "lrz:2023-10-31", "lrz:latest", "lrz:2023-10-31#7", "lrz:2023-10-31/collectionName=main;resourceType=owi"]) - def test_remote_ls_non_empty_lrz(self, runner, temp_dir, specifier): - """Test the 'local' command with a subcommand and specifier.""" - result = runner.invoke(owilix.cli.cli, ['--target', str(temp_dir), 'remote', 'ls', specifier], prog_name='owilix.cli') - assert result.exit_code == 0 - summary = parse_summary_output(result.output) - assert summary["Total Files"]>0 - assert summary["DataCenter lrz"] > 0 - assert summary["resourceType owi"] > 0 - assert summary["Total"] > 0 - - @pytest.mark.parametrize("specifier", ["all", "it4i:2023-12-03", "it4i:latest", "it4i:2023-12-31#7", - "it4i:2023-12-31#7/collectionName=main;resourceType=owi"]) - def test_remote_ls_non_empty_it4i(self, runner, temp_dir, specifier): - """Test the 'local' command with a subcommand and specifier.""" - result = runner.invoke(owilix.cli.cli, ['--target', str(temp_dir), 'remote', 'ls', specifier], - prog_name='owilix.cli') - assert result.exit_code == 0 - summary = parse_summary_output(result.output) - assert summary["Total Files"] > 0 - assert summary["DataCenter it4i"] > 0 - assert summary["resourceType owi"] > 0 - assert summary["Total"] > 0 - - def test_remote_pull_files(self, runner, temp_dir): - """Test the 'remote' command with a subcommand and specifier.""" - result = runner.invoke(owilix.cli.cli, ['--yes', '--target', str(temp_dir), 'remote', 'pull', - 'it4i:2023-12-03', 'files=**/language=slv/*'], prog_name='owilix.cli') - assert result.exit_code == 0 - assert "Fetching files for OWI-Open Web Index-main.owi@it4i-2023-12-3:2023-12-3" in result.output - assert "Found 2 remote files. Syncing with local files" in result.output # Adjust this check based on expected output - result = runner.invoke(owilix.cli.cli, ['--target', str(temp_dir), 'local', 'ls', 'all'], - prog_name='owilix.cli') - assert result.exit_code == 0 - assert len(_fs.ls(str(temp_dir))) == 2 - assert _fs.exists(os.path.join(str(temp_dir), "public")) - assert any([f for f in _fs.ls(os.path.join(str(temp_dir),"public","main")) if f.endswith("json")]) - - - def test_config_command(self, runner, temp_dir): - result = runner.invoke(owilix.cli.cli, ['config', 'set', 'showfields=url', '--target', str(temp_dir)], prog_name='owilix.cli') - assert result.exit_code == 0 - assert "config" in result.output # Adjust this check based on expected output - # Verify that configuration was correctly set in the target directory - config_path = os.path.join(temp_dir, "owilix.cfg") - assert os.path.exists(config_path) - - def test_query_command(self, runner, temp_dir): - """Test the 'query' command with a subcommand and options.""" - result = runner.invoke(owilix.cli.cli, ['query', 'run', '--local', 'all', '--remote', 'lrz', '--target', str(temp_dir)], prog_name='owilix.cli') - assert result.exit_code == 0 - assert "query" in result.output # Adjust this check based on expected output - - def test_clean_command(self, runner, temp_dir): - """Test the 'clean' command.""" - # Pre-create a '.env' file in the temp_dir - env_path = os.path.join(temp_dir, '.env') - with open(env_path, 'w') as f: - f.write('test content') - - result = runner.invoke(owilix.cli.cli, ['clean', '--target', str(temp_dir)], prog_name='owilix.cli') - assert result.exit_code == 0 - assert "Cleaning" in result.output # Check for expected 'clean' command output - - # Verify that the '.env' file has been removed - assert not os.path.exists(env_path) - - def test_logs_command(self, runner, temp_dir): - """Test the 'logs' command with a module argument.""" - result = runner.invoke(owilix.cli.cli, ['--target', str(temp_dir), 'logs', 'lexis', ], prog_name='owilix.cli') - assert result.exit_code == 0 - result = runner.invoke(owilix.cli.cli, ['--target', str(temp_dir), 'logs', 'errors', ], prog_name='owilix.cli') - assert result.exit_code == 0 - assert "log" in result.output - result = runner.invoke(owilix.cli.cli, ['--target', str(temp_dir), 'logs', 'events', ], prog_name='owilix.cli') - assert result.exit_code == 0 - assert "[" in result.output and "]" in result.output - - def test_invalid_command(self, runner): - """Test an invalid command to ensure proper error handling.""" - result = runner.invoke(owilix.cli.cli, ['nonexistent'], prog_name='owilix.cli') - assert result.exit_code != 0 - assert "No such command" in result.output