diff --git a/.gitignore b/.gitignore index 443e702..2d41960 100644 --- a/.gitignore +++ b/.gitignore @@ -1,5 +1,6 @@ # Specific retrorules* +rrparser/cache/ # Generic __pycache__ diff --git a/README.md b/README.md index 1b939ca..cc52744 100644 --- a/README.md +++ b/README.md @@ -7,70 +7,229 @@ Reaction Rules Parser [![Qodana](https://github.com/brsynth/RRParser/actions/workflows/code_quality.yml/badge.svg)](https://github.com/brsynth/RRParser/actions/workflows/code_quality.yml) [![Tests](https://github.com/brsynth/RRParser/actions/workflows/test.yml/badge.svg)](https://github.com/brsynth/RRParser/actions/workflows/test.yml) ## Description -*Reaction Rules Parser*. If no input reaction files is provided, retrieves the reaction rules from [RetroRules](https://retrorules.org). -## Input - -* **rules-file**: (string) Filename of reaction rules -* **outfile**: (string) Filename containing the result of the parsing. If not set, result is printed out in the console. If ends with '.gz', it will be gzipped. -* **input-format**: (string) Valid options: csv, tsv. Format of the input file -* **rule-type**: (string) Valid options: retro, forward, all. Return the rules that are in reverse, forward or both direction -* **diameters**: (integer list) Valid options: 2, 4, 6, 8, 10, 12, 14, 16. The diameter of the rules to return -* **output-format**: (string) Valid options: csv, tar.gz. Format of the returned file +Reaction Rules Parser downloads and caches RetroRules datasets, then writes an +extraction filtered by diameter/radius, rule usage, or EC number. Existing local +CSV/TSV rule files can also be parsed, and rule scores can be replaced. +The default source is **RetroRules 3.1.0 / MetaNetX**, with all even diameters +from **0 to 20** (radii **0 to 10**). The historical **rr02** release remains +available for RetroPath 2.0 workflows. ## Install -### From Conda + +```sh +conda install -c conda-forge rrparser +``` + +## Command line + +### Discover and select rules + +```sh +# List available release/dataset pairs, including rr02 / metanetx. +python -m rrparser --list-releases + +# Use the default release and dataset, including all diameters. +python -m rrparser -o rules.csv + +# Extract Rhea templates with diameters 0, 2, 4, 6 and 8. +python -m rrparser --release 3.1.0 --dataset rhea --diameters 0-8 -o rules.csv + +# Equivalent selection using radii. +python -m rrparser --dataset rhea --radii 0,1,2,3,4 -o rules.csv + +# Historical rr02 rules, restricted to retro usage and diameter 10. +python -m rrparser --release rr02 --rule-type retro -d 10 -o rules.csv + +# Parse an existing local rule file. +python -m rrparser --rules_file input.csv -d 2,4 -o rules.csv +``` + +New releases and their datasets are discovered from the +[RetroRules download page](https://retrorules.org/download), rather than a +hardcoded version list. Available template datasets include `metanetx`, `rhea` +and `uspto`; rr02 supports only `metanetx`. An invalid or unavailable release +reports an error followed by the available release/dataset choices. If the site +is unreachable, the error explains that the list could not be retrieved. + +`--release latest` explicitly selects the newest template release for the +chosen dataset; it does not select rr02. Once downloaded, that selection is +reused from cache, including offline. To update, list available releases and +choose a version explicitly, or delete `retrorules-latest-.txt` from +the cache to resolve `latest` again. The default remains pinned to `3.1.0`. + +### Diameter and radius selection + +`--diameters` accepts comma-separated values, inclusive ranges, and comparisons: + +| Expression | Selected diameters | +| --- | --- | +| `0,4,8` | 0, 4, 8 | +| `0-8` | 0, 2, 4, 6, 8 | +| `<8` | 0, 2, 4, 6 | +| `<=8` | 0, 2, 4, 6, 8 | +| `>8` | 10, 12, 14, 16, 18, 20 | +| `>=8` | 8, 10, 12, 14, 16, 18, 20 | +| `0-4,18,20` | 0, 2, 4, 18, 20 | + +Ranges and comparisons select even values within 0–20. Comma-separated terms +are combined as a union, with duplicates removed. Quote comparisons in the +shell so `<` and `>` are not interpreted as file redirection: + ```sh -[sudo] conda install -c conda-forge rrparser +python -m rrparser --dataset rhea --diameters '<=8' -o rules.csv +python -m rrparser --dataset rhea --diameters '>=8' -o rules.csv ``` -## Use +Alternatively, `--radii` accepts comma-separated integers from 0 to 10: +`--radii 0,1,2` is equivalent to `--diameters 0,2,4`. The two options are +mutually exclusive. Without either option, all diameters 0,2,…,20 are selected. + +For templates, a row is retained when at least one source radius matches +`diameter / 2`, once per template. The output `RADII` column retains all source +radii, separated by semicolons. Explicit odd diameters or values above 20 are +rejected for templates. + +rr02 supplies diameters 2,4,…,16. The extended default therefore selects all +its rules. A selection containing no available diameter, such as `-d 20`, +produces a header-only output after the source has been downloaded or loaded. + +### Source cache and output + +The default cache is **`rrparser/cache/` inside the installed package**, shared +by calls using the same installation. Override it with `--cache-dir` or its +alias `--rules-dir`: + +```sh +python -m rrparser --dataset rhea --cache-dir ./cache -d 0-8 -o rules.csv +``` + +The cache stores complete, original downloads: + +- Template releases: `retrorules--.csv.gz`, for example + `retrorules-3.1.0-rhea.csv.gz`. +- rr02: `retrorules_rr02_rp2_hs.tar.gz`, from the + [historical Zenodo archive](https://zenodo.org/record/5828017/files/retrorules_rr02_rp2_hs.tar.gz). + RRParser reads `retrorules_rr02_rp2_hs/retrorules_rr02_rp2_flat_all.csv` + directly inside the archive without extracting it to disk. + +`-o rules.csv` writes the filtered extraction at the requested location; +filters do not modify the cached source. Changing filters reuses the same +archive without another download. Downloads use a temporary `.part` file, +renamed after completion. The initial rr02 download can take time and currently +has no progress indicator. + +If only an old decompressed rr02 CSV cache exists, the original `.tar.gz` is +downloaded once and subsequently reused. That old CSV is not used by the new +archive cache. + +### Options + +| Option | Purpose / default | +| --- | --- | +| `--rules_file` | Local input file, or `retrorules` (default) to use a cached/downloaded dataset. | +| `--release` | Version, `rr02`, or `latest`; default `3.1.0`. | +| `--dataset` | Dataset name; default `metanetx`. See `--list-releases`. | +| `--list-releases` | Print available release/dataset pairs and exit. | +| `--cache-dir`, `--rules-dir` | Source cache directory; default package-local `rrparser/cache`. | +| `-d`, `--diameters` | List, range, or comparison; default all even diameters 0–20. | +| `--radii` | Comma-separated radii 0–10, as an alternative to `--diameters`. | +| `-rt`, `--rule-type` | `all` (default), `retro`, or `forward`; templates require `all`. | +| `-ec`, `--ec` | File of comma-separated EC prefixes to keep. | +| `-ecx`, `--ecx` | File of comma-separated EC prefixes to exclude; mutually exclusive with `--ec`. | +| `-s`, `--scores` | CSV/TSV file with `Rule ID` and `Score` columns to replace matching scores. | +| `-if`, `--input-format` | Local input format: `csv` (default) or `tsv`. Downloaded rules are CSV. | +| `-of`, `--output-format` | Output format: `csv` (default) or `tsv`. | +| `-o`, `--outfile` | Extraction destination; `.gz` outputs are compressed. Without a destination, results are printed to the console. | +| `--log` | Logging level; default `info`. | +| `--version` | Print RRParser's version and exit. | + +Run `python -m rrparser --help` for the full CLI help. + +## Python API + +Use keyword arguments to select the source, filters and output: -### Function call from Python code ```python -from rrparser import parse_rules - -outfile = parse_rules( - , - , - input_format=<'csv' | 'tsv'>, - rule_type=<'all' | 'retro' | 'forward'>, - diameters=<'2,4,6,8,10,12,14,16'>, - output_format=<'csv' | 'tsv'> +from rrparser import list_retrorules, parse_rules + +# Mapping: (release, dataset) -> download URL. +choices = list_retrorules() + +# Write an extraction. Returns None when outfile is provided. +parse_rules( + outfile="rules.csv", + release="3.1.0", + dataset="rhea", + diameters="<=8", +) + +# Return CSV text instead of writing a file. +csv_text = parse_rules( + outfile=None, + dataset="rhea", + radii="0,1,2", +) + +# Parse a local file with EC filtering and replacement scores. +parse_rules( + outfile="filtered.csv", + rules_file="input.csv", + diameters="2,4", + ec=["1.1"], + scores_file="scores.csv", ) ``` -If parameters from CLI have to be parsed, the function `build_args_parser` is available: +`rules_dir=` overrides the cache directory. `release=`, `dataset=`, and cache +lookup apply when `rules_file='retrorules'`, the default. Diameter expressions +have the same meaning in Python and on the CLI. Pass `radii=` as an alternative +to `diameters=`. `ec=` and `ecx=` take lists of prefixes; `read_ecnumbers()` +can load these from a comma-separated file. Invalid selections raise +`ValueError`; the CLI reports these errors and exits with status 1. + +The CLI argument parser is also available: + ```python from rrparser import build_args_parser -parser = buildparser() -params = parser.parse_args() +parser = build_args_parser() +params = parser.parse_args(["--dataset", "rhea", "--diameters", "<=8"]) ``` -### Run from CLI -```sh -python -m rrparser \ - rules-file \ - [--input-format {csv,tsv}] \ - [--rule-type {all,retro,forward}] \ - [--outfile ] \ - [--diameters {2,4,6,8,10,12,14,16}] \ - [--output-format {csv,tsv}] -``` -If `rules_files` is set to `retrorules`, RetroRules are fetched from `retrorules.org` and considered as input file. +## Output compatibility + +New templates use implicit hydrogens and have a different schema from rr02's +RetroPath 2.0 rules with explicit hydrogens. RRParser renames these columns: + +| Template source column | Output column | +| --- | --- | +| `TEMPLATE_ID` | `Rule ID` | +| `TEMPLATE` | `Rule` | +| `ECS` | `EC number` | +| `SCORE` | `Score` | + +Other source columns, including `REACTIONS`, `RADII`, `RADIUS_MIN`, +`RADIUS_MAX`, and `DATASETS`, are retained. Templates do not provide +retro/forward annotations, so they require `rule_type='all'`. EC filtering and +replacement scores remain supported. rr02 retains its historical output +schema and rule usage filters. Existing local RetroPath CSV/TSV files remain +supported through `--rules_file` or `rules_file=`. ## Tests -Test can be run with the following commands: -### Natively -```bash +Run all tests: + +```sh python -m pytest -v ``` -# CI/CD -For further tests and development tools, a CI toolkit is provided in `ci` folder (see [ci/README.md](ci/README.md)). +Tests use local fixtures, small in-memory archives, and mocked network +responses; no live download is required. They cover discovery, version errors, +cache reuse, original rr02 archive preservation, extraction, diameter/radius +filtering, score replacement, and rr02 retro/forward filters. +CI workflows are available in [`.github/workflows`](.github/workflows). ## Authors diff --git a/environment.yaml b/environment.yaml index b0fb3b5..7d9fc25 100644 --- a/environment.yaml +++ b/environment.yaml @@ -3,6 +3,5 @@ channels: - conda-forge dependencies: - python >=3.6 - - brs_utils - pandas >=1.5.3 - colorlog diff --git a/rrparser/Args.py b/rrparser/Args.py index f902921..004934e 100644 --- a/rrparser/Args.py +++ b/rrparser/Args.py @@ -1,3 +1,4 @@ +# ruff: noqa: N999 -- Preserve the public rrparser.Args import path. """ Created on May 4 2020 @@ -7,13 +8,13 @@ from argparse import ArgumentParser from os import path as os_path + from rrparser._version import __version__ -__PACKAGE_FOLDER = os_path.dirname( - os_path.realpath(__file__) -) -DEFAULT_RULES_DIR = __PACKAGE_FOLDER -DEFAULT_RULES_FILE = 'retrorules' +DEFAULT_RULES_DIR = os_path.join(os_path.dirname(os_path.realpath(__file__)), "cache") +DEFAULT_RULES_FILE = "retrorules" +DEFAULT_RELEASE = "3.1.0" +DEFAULT_DIAMETERS = "0,2,4,6,8,10,12,14,16,18,20" def build_args_parser(): @@ -26,7 +27,9 @@ def build_args_parser(): Argument parser for the rrparser command line interface. """ - parser = ArgumentParser(prog='rrparser', description='Python wrapper to fetch RetroRules') + parser = ArgumentParser( + prog="rrparser", description="Python wrapper to fetch RetroRules" + ) parser = _add_arguments(parser) return parser @@ -35,67 +38,124 @@ def _add_arguments(parser): # Positional arguments parser.add_argument( - '--rules_file', # must be '_' otherwise it will be touchy in 'args' Namespace + "--rules_file", # must be '_' otherwise it will be touchy in 'args' Namespace type=str, default=DEFAULT_RULES_FILE, - help="rules file to parse. If set to 'retrorules'," \ - + " RetroRules are considered as input file, either locally or fetched over Internet." + help="rules file to parse. If set to 'retrorules'," + + " RetroRules are considered as input file, either locally or fetched over Internet.", ) # Optional arguments - parser.add_argument('-o', '--outfile', - type=str, - help="file where results are written. If file ends with '.gz', it will be gzipped.") - parser.add_argument('-if', '--input-format', - type=str, - choices=['csv', 'tsv'], - default='csv', - help='input file format (default: csv)') - parser.add_argument('-rt', '--rule-type', - type=str, - choices=['all', 'retro', 'forward'], - default='all', - help="rule usage to filter from rules file") - parser.add_argument('-d', '--diameters', - type=str, - default='2,4,6,8,10,12,14,16', - help='diameter of the sphere including the atoms around " \ - + "the reacting center (default is including all values: 2,4,6,8,10,12,14,16)." \ - + " The higher is the diameter, the more specific are the rules') - parser.add_argument('-of', '--output-format', - type=str, - choices=['csv', 'tsv'], - default='csv', - help='output file format (default: csv)') + parser.add_argument( + "--list-releases", + action="store_true", + help="List releases and datasets available on retrorules.org, then exit", + ) + parser.add_argument( + "--release", + default=DEFAULT_RELEASE, + help="RetroRules version, latest or rr02 (default: 3.1.0)", + ) + parser.add_argument( + "--dataset", + default="metanetx", + help="RetroRules dataset (default: metanetx; see --list-releases)", + ) + parser.add_argument( + "-o", + "--outfile", + type=str, + help="file where results are written. If file ends with '.gz', it will be gzipped.", + ) + parser.add_argument( + "-if", + "--input-format", + type=str, + choices=["csv", "tsv"], + default="csv", + help="input file format (default: csv)", + ) + parser.add_argument( + "-rt", + "--rule-type", + type=str, + choices=["all", "retro", "forward"], + default="all", + help="rule usage to filter from rules file", + ) + sizes = parser.add_mutually_exclusive_group() + sizes.add_argument( + "-d", + "--diameters", + type=str, + default=DEFAULT_DIAMETERS, + help='Diameters: list (0,2,4), inclusive range (0-8), "<8", "<=8", ">8" or ">=8". ' + "Ranges and comparisons select even values from 0 to 20 (default: all). " + "For templates, diameter = 2 * radius; rr02 provides diameters 2 to 16", + ) + sizes.add_argument( + "--radii", + type=str, + help="Comma-separated radii from 0 to 10 (alternative to --diameters)", + ) + parser.add_argument( + "-of", + "--output-format", + type=str, + choices=["csv", "tsv"], + default="csv", + help="output file format (default: csv)", + ) # Options --ecx and --ec are mutually exclusive group = parser.add_mutually_exclusive_group() - group.add_argument('-ecx', '--ecx', - type=str, - help='file containing EC numbers to filter out') - group.add_argument('-ec', '--ec', - type=str, - help='file containing EC numbers to filter in') + group.add_argument( + "-ecx", "--ecx", type=str, help="file containing EC numbers to filter out" + ) + group.add_argument( + "-ec", "--ec", type=str, help="file containing EC numbers to filter in" + ) # Option to customize rules score with a dedicated file (CSV or TSV) # where each line is the rule name and its score - parser.add_argument('-s', '--scores', - type=str, - help='CSV or TSV file containing rules names and scores') + parser.add_argument( + "-s", + "--scores", + type=str, + help="CSV or TSV file containing rules names and scores", + ) # Program options - parser.add_argument('--log', metavar='ARG', - type=str, choices=['debug', 'info', 'warning', 'error', 'critical', - 'DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'], - default='def_info', - help='Adds a console logger for the specified level (default: error)') - parser.add_argument('--version', action='version', - version='%(prog)s {}'.format(__version__), - help='show the version number and exit') parser.add_argument( - '--rules-dir', - default=__PACKAGE_FOLDER, - help='Path to the rules directory (default in package directory).' + "--log", + metavar="ARG", + type=str, + choices=[ + "debug", + "info", + "warning", + "error", + "critical", + "DEBUG", + "INFO", + "WARNING", + "ERROR", + "CRITICAL", + ], + default="def_info", + help="Adds a console logger for the specified level (default: error)", + ) + parser.add_argument( + "--version", + action="version", + version=f"%(prog)s {__version__}", + help="show the version number and exit", + ) + parser.add_argument( + "--rules-dir", + "--cache-dir", + default=DEFAULT_RULES_DIR, + help="Source archive cache (default: %(default)s). Extracted rules are written to --outfile.", ) return parser diff --git a/rrparser/Downloads.py b/rrparser/Downloads.py new file mode 100644 index 0000000..ee02f2a --- /dev/null +++ b/rrparser/Downloads.py @@ -0,0 +1,134 @@ +# ruff: noqa: N999 -- Preserve the public rrparser.Downloads import path. +"""Discover CSV datasets on RetroRules and cache them locally.""" + +from html.parser import HTMLParser +from pathlib import Path +from re import fullmatch +from shutil import copyfileobj +from urllib.error import URLError +from urllib.parse import parse_qs, urljoin, urlsplit +from urllib.request import urlopen + +from .Args import DEFAULT_RELEASE + +DOWNLOAD_PAGE = "https://retrorules.org/download" +RR02_URL = "https://zenodo.org/record/5828017/files/retrorules_rr02_rp2_hs.tar.gz" +RR02_FILE = "retrorules_rr02_rp2_hs/retrorules_rr02_rp2_flat_all.csv" +RR02_CACHE_FILE = "retrorules_rr02_rp2_hs.tar.gz" + + +class _DatasetLinks(HTMLParser): + def __init__(self): + super().__init__() + self.datasets = {} + + def handle_starttag(self, tag, attrs): + if tag != "a": + return + url = urljoin(DOWNLOAD_PAGE, dict(attrs).get("href", "")) + parts = urlsplit(url) + match = fullmatch(r"/dl/v(\d+\.\d+(?:\.\d+)?)/([^/]+)/templates", parts.path) + if ( + parts.netloc == "retrorules.org" + and match + and parse_qs(parts.query).get("format") == ["csv"] + ): + self.datasets[match.groups()] = url + + +def list_retrorules(): + """Return {(release, dataset): download URL}, newest releases first.""" + parser = _DatasetLinks() + try: + with urlopen(DOWNLOAD_PAGE, timeout=30) as response: + parser.feed(response.read().decode("utf-8")) + except URLError as exc: + raise ValueError(f"Cannot discover RetroRules datasets: {exc}") from exc + if not parser.datasets: + raise ValueError(f"No CSV datasets found on {DOWNLOAD_PAGE}") + datasets = dict( + sorted( + parser.datasets.items(), + key=lambda item: tuple(map(int, item[0][0].split("."))), + reverse=True, + ) + ) + datasets[("rr02", "metanetx")] = RR02_URL + return datasets + + +def _download(url, filename): + """Cache the original download without extracting or recompressing it.""" + if filename.is_file(): + return str(filename) + filename.parent.mkdir(parents=True, exist_ok=True) + temporary = filename.with_suffix(".part") + try: + with urlopen(url, timeout=30) as response, temporary.open("wb") as output: + copyfileobj(response, output) + temporary.replace(filename) + except URLError as exc: + raise ValueError(f"Cannot download RetroRules: {exc}") from exc + finally: + if temporary.exists(): + temporary.unlink() + return str(filename) + + +def _selection_error(message, datasets=None): + if datasets is None: + try: + datasets = list_retrorules() + except ValueError as exc: + return ValueError( + f"{message}\nAvailable releases could not be retrieved: {exc}" + ) + choices = "\n".join(f"{version}\t{name}" for version, name in datasets) + return ValueError(f"{message}\nAvailable releases and datasets:\n{choices}") + + +def fetch_dataset(rules_dir, release=DEFAULT_RELEASE, dataset="metanetx"): + """Reuse cached rules; resolve latest only on the first download.""" + if release == "rr02": + if dataset != "metanetx": + raise ValueError("RetroRules rr02 is only available for metanetx") + return _download(RR02_URL, Path(rules_dir) / RR02_CACHE_FILE) + latest = release == "latest" + if not latest and not fullmatch(r"\d+\.\d+(?:\.\d+)?", release): + raise _selection_error( + f"Invalid release {release!r}; use latest, rr02 or a version such as 3.1.0" + ) + if not fullmatch(r"[a-z0-9_-]+", dataset): + raise ValueError("Invalid RetroRules dataset name") + cache = Path(rules_dir) + selection = cache / (f"retrorules-latest-{dataset}.txt") + if release == "latest" and selection.is_file(): + saved = selection.read_text().strip() + if fullmatch(r"\d+\.\d+(?:\.\d+)?", saved): + cached = cache / (f"retrorules-{saved}-{dataset}.csv.gz") + if cached.is_file(): + return str(cached) + filename = cache / (f"retrorules-{release}-{dataset}.csv.gz") + if release != "latest" and filename.is_file(): + return str(filename) + datasets = list_retrorules() + if release == "latest": + release = next( + ( + version + for version, name in datasets + if name == dataset and version != "rr02" + ), + None, + ) + url = datasets.get((release, dataset)) + if url is None: + raise _selection_error( + f"Release/dataset unavailable: {release} / {dataset}", datasets + ) + cache.mkdir(parents=True, exist_ok=True) + filename = cache / (f"retrorules-{release}-{dataset}.csv.gz") + _download(url, filename) + if latest: + selection.write_text(release) + return str(filename) diff --git a/rrparser/Parser.py b/rrparser/Parser.py index 8fb8943..f415ba7 100644 --- a/rrparser/Parser.py +++ b/rrparser/Parser.py @@ -1,3 +1,4 @@ +# ruff: noqa: N999 -- Preserve the public rrparser.Parser import path. """ Created on May 4 2020 @@ -5,42 +6,27 @@ """ -from os import ( - path as os_path -) -from brs_utils import ( - download_and_extract_tar_gz -) -from pandas import ( - read_csv, - DataFrame -) -from csv import ( - QUOTE_ALL, - QUOTE_NONE -) -from typing import ( - List, - Tuple, - Literal -) -from logging import ( - Logger, - getLogger -) +from __future__ import annotations + +from csv import QUOTE_ALL, QUOTE_NONE +from logging import Logger, getLogger +from re import fullmatch +from tarfile import open as open_tar + +from pandas import DataFrame, read_csv + from .Args import ( + DEFAULT_DIAMETERS, + DEFAULT_RELEASE, + DEFAULT_RULES_DIR, DEFAULT_RULES_FILE, - DEFAULT_RULES_DIR ) +from .Downloads import RR02_FILE, fetch_dataset +LOGGER = getLogger(__name__) -RETRORULES_URL = 'https://zenodo.org/record/5828017/files/retrorules_rr02_rp2_hs.tar.gz' - -def read_ecnumbers( - ec_file: str, - logger: Logger = getLogger(__name__) -) -> List[str]: +def read_ecnumbers(ec_file: str, logger: Logger = LOGGER) -> list[str]: """ Read EC numbers from a file. @@ -56,25 +42,28 @@ def read_ecnumbers( ecx: List[str] List of EC numbers. """ - logger.debug('Reading EC numbers from file...') - with open(ec_file, 'r') as f: - ec = f.read().split(',') + logger.debug("Reading EC numbers from file...") + with open(ec_file, "r") as f: + ec = f.read().split(",") return ec def parse_rules( - outfile: str, - rules_file: str = DEFAULT_RULES_FILE, - scores_file: str = None, - rules_dir: str = DEFAULT_RULES_DIR, - input_format: str = 'csv', - rule_type: str = 'all', - diameters: str = '2,4,6,8,10,12,14,16', - ecx: List[str] = [], - ec: List[str] = [], - output_format: str = 'csv', - logger: Logger = getLogger(__name__) -) -> None or str: + outfile: str | None, + rules_file: str = DEFAULT_RULES_FILE, + scores_file: str | None = None, + rules_dir: str = DEFAULT_RULES_DIR, + input_format: str = "csv", + rule_type: str = "all", + diameters: str = DEFAULT_DIAMETERS, + ecx: list[str] | None = None, + ec: list[str] | None = None, + output_format: str = "csv", + logger: Logger = LOGGER, + release: str = DEFAULT_RELEASE, + dataset: str = "metanetx", + radii: str | None = None, +) -> str | None: """ Parse a reaction rules file and extract sub-part according 'diameters' and 'rule_type' filters. @@ -85,7 +74,7 @@ def parse_rules( scores_file: str Reactions scores file (if not set RetroRules are considered) rules_dir: str - Directory where to store RetroRules (if not set RetroRules are stored in the current directory) + Source archive cache (default: cache directory inside the rrparser package). outfile: str Filename where to write results (if not set results are returned as a string) input_format: str @@ -93,7 +82,7 @@ def parse_rules( rule_type: str Type of reaction rules ['all', 'forward', 'retro'] (default 'all') diameters: str - Diameters to filter [2,4,6,8,10,12,14,16] (default) + Diameters to filter (default: all even values from 0 to 20). ecx: List[str] List of EC numbers to remove from rules ec: List[str] @@ -102,6 +91,12 @@ def parse_rules( Format of file results are written into logger : Logger The logger object. + release: str + RetroRules version (default 3.1.0), rr02, or latest. + dataset: str + Dataset discovered on retrorules.org (default metanetx). + radii: str + Comma-separated radii 0–10, as an alternative to diameters. Returns ------- @@ -110,48 +105,77 @@ def parse_rules( """ # Check args is here since the method is directly callable. + if radii is not None: + if diameters != DEFAULT_DIAMETERS: + raise ValueError("Use either radii or diameters") + values = radii.split(",") + if any(not value.isdigit() or int(value) > 10 for value in values): + raise ValueError("Radii must be integers from 0 to 10, separated by commas") + diameters = ",".join(str(2 * int(value)) for value in values) diameters, sep, quoting = check_args( - rule_type, - diameters, - input_format, - output_format + rule_type, diameters, input_format, output_format ) # If 'rules_file' is set to RetroRules, then fetch RetroRules on Internet - if rules_file == 'retrorules': - rules_file = fetch_retrorules(rules_dir) + if rules_file == "retrorules": + rules_file = fetch_retrorules(rules_dir, logger, release, dataset) + sep = "," - logger.debug('Reading values...') + logger.debug("Reading values...") # Read 'csv' as 'tsv' by specying separator - rf = read_csv( - rules_file, - sep=sep, - float_precision='round_trip' - ) + read_options = { + "sep": sep, + "dtype": {"RADII": str}, + "float_precision": "round_trip", + } + if str(rules_file).endswith(".tar.gz"): + with open_tar(rules_file, mode="r:gz") as archive: + member = next( + ( + item + for item in archive + if item.name in (RR02_FILE, "./" + RR02_FILE) and item.isfile() + ), + None, + ) + if member is None: + raise ValueError( + f"RetroRules rr02 archive does not contain {RR02_FILE}" + ) + with archive.extractfile(member) as source: + rf = read_csv(source, **read_options) + else: + rf = read_csv(rules_file, **read_options) # Filter rules according to 'rule_type' and 'diameters' + templates = "TEMPLATE_ID" in rf.columns + if templates: + rf = rf.rename( + columns={ + "TEMPLATE_ID": "Rule ID", + "TEMPLATE": "Rule", + "ECS": "EC number", + "SCORE": "Score", + } + ) results = filter_(rf, rule_type, diameters, ecx, ec, logger=logger) # If 'rules_scores_file' is set, then read it and merge with 'results' if scores_file: - logger.debug('Reading scores values...') - rsf = read_csv( - scores_file, - sep=sep, - float_precision='round_trip' - ) + logger.debug("Reading scores values...") + rsf = read_csv(scores_file, sep=sep, float_precision="round_trip") results = replace_scores(results, rsf, logger=logger) - logger.info('Writing results...') + logger.info("Writing results...") return results.to_csv( outfile, index=False, - sep=sep, - quoting=quoting + sep=("," if output_format == "csv" else "\t") if templates else sep, + **({} if templates else {"quoting": quoting}), ) -def replace_scores(results, scores, logger: Logger = getLogger(__name__)): +def replace_scores(results, scores, logger: Logger = LOGGER): """ Replace scores in 'results' with those in 'rules_scores_file'. @@ -167,14 +191,14 @@ def replace_scores(results, scores, logger: Logger = getLogger(__name__)): The logger object. """ # Replace scores in results with those in 'rules_scores_file' for the same rules - logger.debug('Merging scores with rules...') + logger.debug("Merging scores with rules...") # Create a mapping from df2 - score_map = scores.set_index('Rule ID')['Score'] + score_map = scores.set_index("Rule ID")["Score"] # Update the Score in df1 if Rule ID is in df2 results = results.copy() # warning if Rule ID has been found # in scores but not in results - missing_rules = set(scores['Rule ID']) - set(results['Rule ID']) + missing_rules = set(scores["Rule ID"]) - set(results["Rule ID"]) if missing_rules: logger.warning( f"Rule IDs {missing_rules} found in scores file but not in results. " @@ -182,8 +206,8 @@ def replace_scores(results, scores, logger: Logger = getLogger(__name__)): ) # Update the Score in results DataFrame # using the mapping from scores - results['Score'] = results['Rule ID'].map(score_map).fillna(results['Score']) - results.loc[:, 'Score'] = results['Rule ID'].map(score_map).fillna(results['Score']) + results["Score"] = results["Rule ID"].map(score_map).fillna(results["Score"]) + results.loc[:, "Score"] = results["Rule ID"].map(score_map).fillna(results["Score"]) # # Optional: convert Score back to the original dtype if needed # results['Score'] = results['Score'].astype(float) return results @@ -192,11 +216,11 @@ def replace_scores(results, scores, logger: Logger = getLogger(__name__)): def filter_( df: DataFrame, rule_type: str, - diameters: List[int], - ecx: List[str] = [], - ec: List[str] = [], - logger: Logger = getLogger(__name__) -) -> List[int]: + diameters: list[int], + ecx: list[str] | None = None, + ec: list[str] | None = None, + logger: Logger = LOGGER, +) -> DataFrame: """ Filter a pandas dataframe with 'Diameters' and 'Rule usage' criteria. @@ -220,48 +244,67 @@ def filter_( query: Pandas dataframe Original dataframe filtered according to 'rule_type' and 'diameters'. """ - logger.debug( - 'Args: {df}, {rt}, {dia}'.format( - df=df, - rt=rule_type, - dia=diameters, - ecx=ecx - ) - ) + logger.debug(f"Args: {df}, {rule_type}, {diameters}") + ecx = ecx if ecx is not None else [] + ec = ec if ec is not None else [] ec_filter = ecx if ecx else ec - filter = 'in' if ec else 'not in' - logger.info('Filtering rules:') - logger.info(f' -> EC numbers ({filter}): {ec_filter}') - logger.info(f' -> Diameters: {diameters}') - logger.info(f' -> Rule types: {rule_type}') + filter = "in" if ec else "not in" + logger.info("Filtering rules:") + logger.info(f" -> EC numbers ({filter}): {ec_filter}") + logger.info(f" -> Diameters: {diameters}") + logger.info(f" -> Rule types: {rule_type}") # Identify rules that contain at least one EC number starting by a pattern contained in 'ecx' rules_idx = filter_rules_by_ec(df, ec_filter, logger) # Build the filtering query query = f"index {filter} @rules_idx" - query += ' & Diameter == @diameters' - if rule_type != 'all': - query += ' & `Rule usage` == @rule_usage_filter' - logger.debug(f'Query: {query}') - # Used in 'e' in case of exception raised - rule_usage_filter = ['both', rule_type] + query_vars = { + "rules_idx": rules_idx, + "diameters": diameters, + "rule_usage_filter": ["both", rule_type], + } + if "RADII" in df.columns: + if any(diameter % 2 or diameter > 20 for diameter in diameters): + raise ValueError( + "Template diameters must be even values from 0 to 20 (radii 0 to 10)" + ) + if rule_type != "all": + raise ValueError( + "RetroRules templates have no retro/forward annotation; use --rule-type all" + ) + radii = {diameter // 2 for diameter in diameters} + radius_idx = df[ + df["RADII"] + .fillna("") + .astype(str) + .apply( + lambda values: any( + int(radius) in radii for radius in values.split(";") if radius + ) + ) + ].index + query_vars["radius_idx"] = radius_idx + query += " & index in @radius_idx" + else: + query += " & Diameter == @diameters" + if rule_type != "all": + query += " & `Rule usage` == @rule_usage_filter" + logger.debug(f"Query: {query}") try: - _df = df.query(query) - logger.info('') - logger.info(f'=> {len(_df)} rules found') - logger.info('') + _df = df.query(query, local_dict=query_vars) + logger.info("") + logger.info(f"=> {len(_df)} rules found") + logger.info("") return _df - except Exception as e: - raise KeyError(e) + except (KeyError, TypeError, ValueError) as e: + raise KeyError(e) from e def filter_rules_by_ec( - df: DataFrame, - ec_filter: List[str], - logger: Logger = getLogger(__name__) -) -> List[int]: + df: DataFrame, ec_filter: list[str], logger: Logger = LOGGER +) -> list[int]: """ Filter out rules that contain at least one EC number that starts with one in 'ecx'. @@ -275,19 +318,23 @@ def filter_rules_by_ec( The logger object. """ # Consider EC numbers as list (separator: ';') - ec_numbers = df['EC number'].str.split(';') + ec_numbers = df["EC number"].fillna("").str.split(";") # Find indices of rows where any EC number starts with one in 'ec_filter' def filter_ec_numbers(ec_list, ec_filter): return any(any(ec.startswith(ec_f) for ec_f in ec_filter) for ec in ec_list) # Apply filtering function and get indices - rules_idx = df[ec_numbers.apply(lambda x: filter_ec_numbers(x, ec_filter))].index.tolist() + rules_idx = df[ + ec_numbers.apply(lambda x: filter_ec_numbers(x, ec_filter)) + ].index.tolist() # Debug log if logger.level == 10: for i in rules_idx: - logger.debug(f"Tagging rule {i} with EC numbers {df.at[i, 'EC number']} because an EC number appears in the exclusion filter") + logger.debug( + f"Tagging rule {i} with EC numbers {df.at[i, 'EC number']} because an EC number appears in the exclusion filter" + ) # # Remove rules that contain at least one EC number that starts with one in 'ecx', # # e.g. rule with EC numbers = ['1.1.1.1', '2.2.2.2'] will be removed if 'ecx' contains '1.1' @@ -303,7 +350,9 @@ def filter_ec_numbers(ec_list, ec_filter): def fetch_retrorules( rules_dir: str = DEFAULT_RULES_DIR, - logger: Logger = getLogger(__name__) + logger: Logger = LOGGER, + release: str = DEFAULT_RELEASE, + dataset: str = "metanetx", ) -> str: """ Fetch RetroRules over Internet if not already on disk. @@ -311,7 +360,7 @@ def fetch_retrorules( Parameters ---------- rules_dir: str - Directory where to store RetroRules (if not set RetroRules are stored in the current directory) + Directory where to cache RetroRules source files. logger : Logger The logger object. @@ -320,18 +369,37 @@ def fetch_retrorules( filename: str Path to downoaded filename. """ - filename = 'retrorules_rr02_rp2_hs/retrorules_rr02_rp2_flat_all.csv' - rules_file = os_path.join(rules_dir, filename) - - if not os_path.exists(rules_file): - logger.info('Downloading retrorules file...') - download_and_extract_tar_gz( - RETRORULES_URL, - rules_dir, - filename - ) - - return rules_file + return fetch_dataset(rules_dir, release, dataset) + + +def parse_diameters(expression: str) -> list[int]: + """Expand lists, inclusive ranges and comparisons into diameters.""" + diameters = set() + for term in expression.split(","): + term = term.strip() + interval = fullmatch(r"(\d+)\s*-\s*(\d+)", term) + comparison = fullmatch(r"([<>])(=?)\s*(\d+)", term) + if term.isdigit(): + diameters.add(int(term)) + elif interval: + lower, upper = map(int, interval.groups()) + if lower > upper: + raise ValueError("Diameter range must have its lower bound first") + diameters.update(d for d in range(0, 21, 2) if lower <= d <= upper) + elif comparison: + operator, inclusive, bound = comparison.groups() + bound = int(bound) + diameters.update( + d + for d in range(0, 21, 2) + if (d < bound if operator == "<" else d > bound) + or (inclusive and d == bound) + ) + else: + raise ValueError( + "Invalid --diameters: use a list (0,2,4), range (0-8), <8, <=8, >8 or >=8" + ) + return sorted(diameters) def check_args( @@ -339,8 +407,8 @@ def check_args( diameters: str, input_format: str, output_format: str, - logger: Logger = getLogger(__name__) -) -> Tuple[List[int], str, str]: + logger: Logger = LOGGER, +) -> tuple[list[int], str, int]: """ Check arguments are well-formed and format them. @@ -349,7 +417,7 @@ def check_args( rule_type: str Type of reaction rules ['all', 'forward', 'retro'] (default 'all') diameters: str - Diameters to filter [2,4,6,8,10,12,14,16] (default) + Diameters to filter (default: all even values from 0 to 20). input_format: str Input format of rules_file ['csv', 'tsv'] (default 'csv') output_format: str @@ -364,40 +432,29 @@ def check_args( sep - separation character detected from 'input_format'. quoting - quoting behavior detected from 'output_format'. """ - logger.debug( - 'Args: {rt}, {dia}, {i_f}, {o_f}'.format( - rt=rule_type, - dia=diameters, - i_f=input_format, - o_f=output_format - ) - ) + logger.debug(f"Args: {rule_type}, {diameters}, {input_format}, {output_format}") # Rule type - if not rule_type or rule_type not in ['all', 'retro', 'forward']: - raise ValueError('Cannot detect \'rule_type\' input: '+str(rule_type)) + if not rule_type or rule_type not in ["all", "retro", "forward"]: + raise ValueError("Cannot detect 'rule_type' input: " + str(rule_type)) # Diameter - diameters = diameters.split(',') - for d in diameters: - if not d.isdigit(): - raise ValueError( - "--diameters takes only digit separated by comma") + diameters = parse_diameters(diameters) # Input format - if input_format == 'csv': - sep = ',' - elif input_format == 'tsv': - sep = '\t' + if input_format == "csv": + sep = "," + elif input_format == "tsv": + sep = "\t" else: - raise ValueError('Can only have \'csv\' or \'tsv\' input formats') + raise ValueError("Can only have 'csv' or 'tsv' input formats") # Output format - if output_format == 'csv': + if output_format == "csv": quoting = QUOTE_NONE - elif output_format == 'tsv': + elif output_format == "tsv": quoting = QUOTE_ALL else: - raise ValueError('Can only have \'csv\' or \'tsv\' input formats') + raise ValueError("Can only have 'csv' or 'tsv' input formats") - return list(map(int, diameters)), sep, quoting + return diameters, sep, quoting diff --git a/rrparser/__init__.py b/rrparser/__init__.py index 5ddb534..25abe83 100644 --- a/rrparser/__init__.py +++ b/rrparser/__init__.py @@ -5,8 +5,15 @@ """ -from rrparser.Parser import parse_rules, read_ecnumbers -from rrparser.Args import build_args_parser from rrparser._version import __version__ +from rrparser.Args import build_args_parser +from rrparser.Downloads import list_retrorules +from rrparser.Parser import parse_rules, read_ecnumbers -__all__ = ["parse_rules", "build_args_parser"] +__all__ = [ + "__version__", + "build_args_parser", + "list_retrorules", + "parse_rules", + "read_ecnumbers", +] diff --git a/rrparser/__main__.py b/rrparser/__main__.py index dece697..2da2f46 100644 --- a/rrparser/__main__.py +++ b/rrparser/__main__.py @@ -1,43 +1,39 @@ -#!/usr/bin/env python - """ rrparser command line interface. """ -from rrparser import ( - build_args_parser, - parse_rules, - read_ecnumbers -) + +from logging import Logger, StreamHandler, getLogger + from colorlog import ColoredFormatter -from logging import ( - Logger, - getLogger, - StreamHandler -) +from rrparser import build_args_parser, parse_rules, read_ecnumbers from rrparser._version import __version__ +from rrparser.Downloads import list_retrorules def entry_point(): """ Entry point for the rrparser command line interface. - + """ parser = build_args_parser() - args = parser.parse_args() + args = parser.parse_args() # Create logger logger = create_logger(parser.prog, args.log) - logger.info( - '\n{prog} {version}\n'.format( - prog=logger.name, - version=__version__ - ) - ) + logger.info(f"\n{logger.name} {__version__}\n") logger.debug(args) + if args.list_releases: + try: + for release, dataset in list_retrorules(): + print(f"{release}\t{dataset}") + except ValueError as exc: + parser.exit(1, f"{exc}\n") + return + # Read 'ecx' from file # EC numbers are separated by a comma ecx = [] @@ -59,7 +55,10 @@ def entry_point(): ecx=ecx, ec=ec, output_format=args.output_format, - logger=logger + logger=logger, + release=args.release, + dataset=args.dataset, + radii=args.radii, ) # Print results if 'outfile' is empty @@ -67,13 +66,10 @@ def entry_point(): logger.info(results) except ValueError as e: - logger.error(str(e)) + parser.exit(1, f"{e}\n") -def create_logger( - name: str = __name__, - log_level: str = 'def_info' -) -> Logger: +def create_logger(name: str = __name__, log_level: str = "def_info") -> Logger: """ Create a logger with name and log_level. @@ -90,19 +86,20 @@ def create_logger( Logger The logger object. - """ + """ logger = getLogger(name) handler = StreamHandler() - if log_level.startswith('def_'): - log_format = '%(log_color)s%(message)s%(reset)s' + if log_level.startswith("def_"): + log_format = "%(log_color)s%(message)s%(reset)s" log_level = log_level[4:] else: - log_format = \ - '%(log_color)s%(levelname)-8s' \ - + ' | %(asctime)s.%(msecs)03d %(module)s' \ - + ' - %(funcName)s(): %(message)s%(reset)s' - + log_format = ( + "%(log_color)s%(levelname)-8s" + + " | %(asctime)s.%(msecs)03d %(module)s" + + " - %(funcName)s(): %(message)s%(reset)s" + ) + formatter = ColoredFormatter(log_format) handler.setFormatter(formatter) logger.addHandler(handler) @@ -111,5 +108,5 @@ def create_logger( return logger -if __name__ == '__main__': +if __name__ == "__main__": entry_point() diff --git a/setup.py b/setup.py index 2e4bff6..87d8158 100644 --- a/setup.py +++ b/setup.py @@ -1,40 +1,36 @@ """Setup script for conda package""" -from setuptools import setup + from os import path as os_path +from setuptools import setup # INFOS -package = 'rrparser' -descr = 'Reaction Rules Parser' -url = 'https://github.com/brsynth/rrparser' -authors = 'Joan Hérisson, Melchior du Lac, Thomas Duigou' -corr_author = 'joan.herisson@univ-evry.fr' +package = "rrparser" +descr = "Reaction Rules Parser" +url = "https://github.com/brsynth/rrparser" +authors = "Joan Hérisson, Melchior du Lac, Thomas Duigou" +corr_author = "joan.herisson@univ-evry.fr" # LONG DESCRIPTION with open( - os_path.join( - os_path.dirname(os_path.realpath(__file__)), - 'README.md' - ), - 'r', - encoding='utf-8' + os_path.join(os_path.dirname(os_path.realpath(__file__)), "README.md"), + "r", + encoding="utf-8", ) as f: long_description = f.read() + # VERSION def get_version(): """Get version from CHANGELOG.md""" with open( - os_path.join( - os_path.dirname(os_path.realpath(__file__)), - 'CHANGELOG.md' - ), - 'r' + os_path.join(os_path.dirname(os_path.realpath(__file__)), "CHANGELOG.md"), "r" ) as f1: lines = f1.readlines() for line in lines: - if line.startswith('##'): + if line.startswith("##"): from re import search + m = search(r"\[(.+)]", line) if m: return m.group(1) @@ -47,18 +43,17 @@ def get_version(): author_email=corr_author, description=descr, long_description=long_description, - long_description_content_type='text/markdown', + long_description_content_type="text/markdown", url=url, packages=[package], package_dir={package: package}, include_package_data=True, - test_suite='pytest', - license='MIT', + test_suite="pytest", + license="MIT", classifiers=[ - 'Programming Language :: Python :: 3', - 'License :: OSI Approved :: MIT License', - 'Operating System :: OS Independent', + "Programming Language :: Python :: 3", + "License :: OSI Approved :: MIT License", + "Operating System :: OS Independent", ], - python_requires='>=3.7' + python_requires=">=3.7", ) - diff --git a/tests/test_diameter_expressions.py b/tests/test_diameter_expressions.py new file mode 100644 index 0000000..42b5265 --- /dev/null +++ b/tests/test_diameter_expressions.py @@ -0,0 +1,76 @@ +from io import StringIO +from pathlib import Path + +import pandas as pd +import pytest + +from rrparser import build_args_parser, parse_rules +from rrparser.Parser import parse_diameters + + +@pytest.mark.parametrize( + "expression,expected", + [ + ("0,2,4", [0, 2, 4]), + ("0-8", [0, 2, 4, 6, 8]), + ("<8", [0, 2, 4, 6]), + ("<=8", [0, 2, 4, 6, 8]), + (">8", [10, 12, 14, 16, 18, 20]), + (">=8", [8, 10, 12, 14, 16, 18, 20]), + (">20", []), + (">=20", [20]), + (">=0", list(range(0, 21, 2))), + (">9", [10, 12, 14, 16, 18, 20]), + (">= 18", [18, 20]), + ("<4,>=18", [0, 2, 18, 20]), + ("0-8,20,8", [0, 2, 4, 6, 8, 20]), + ("1-9", [2, 4, 6, 8]), + (" 0 - 8 ", [0, 2, 4, 6, 8]), + ("<=0", [0]), + ("<0", []), + ("18-20", [18, 20]), + ("<=20", list(range(0, 21, 2))), + ], +) +def test_expressions(expression, expected, tmp_path): + assert parse_diameters(expression) == expected + args = build_args_parser().parse_args(["--diameters", expression]) + rules = tmp_path / "templates.csv" + pd.DataFrame( + { + "TEMPLATE_ID": [f"RR:{radius}" for radius in range(11)], + "TEMPLATE": ["[C]>>[O]"] * 11, + "SCORE": [1.0] * 11, + "ECS": ["1.1.1.1"] * 11, + "RADII": list(range(11)), + } + ).to_csv(rules, index=False) + output = parse_rules(None, rules_file=str(rules), diameters=args.diameters) + assert pd.read_csv(StringIO(output))["Rule ID"].tolist() == [ + f"RR:{diameter // 2}" for diameter in expected + ] + + +@pytest.mark.parametrize( + "expression", ["", "8-0", "2-", "<=x", ">=x", ">>8", "0,,2", "-2"] +) +def test_invalid_expressions(expression): + with pytest.raises(ValueError): + parse_diameters(expression) + + +@pytest.mark.parametrize( + "expression,enumeration", + [ + ("0-8", "0,2,4,6,8"), + ("<8", "0,2,4,6"), + ("<=8", "0,2,4,6,8"), + (">8", "10,12,14,16,18,20"), + (">=8", "8,10,12,14,16,18,20"), + ], +) +def test_legacy_rules(expression, enumeration): + rules = str(Path(__file__).parent / "data/rules.csv") + assert parse_rules(None, rules_file=rules, diameters=expression) == parse_rules( + None, rules_file=rules, diameters=enumeration + ) diff --git a/tests/test_diameters.py b/tests/test_diameters.py index 77f5259..63859a1 100644 --- a/tests/test_diameters.py +++ b/tests/test_diameters.py @@ -1,68 +1,24 @@ -""" -Created on June 17 2020 +from pathlib import Path +from tempfile import TemporaryDirectory -@author: Joan Hérisson -""" - -from os import unlink - -# Generic for test process from test_main import Test_RR -# Specific for tool from rrparser.Parser import parse_rules -# Specific for tests themselves -from io import open as io_open -from tempfile import NamedTemporaryFile - - -# Cette classe est un groupe de tests. Son nom DOIT commencer -# par 'Test' et la classe DOIT hériter de unittest.TestCase. -# 'Test_' prefix is mandatory class Test_RR_Diameters(Test_RR): - - def test_BadDiametersArgument(self): - for diam in ['3']: - with self.subTest(diam=diam): - outfile = NamedTemporaryFile(delete=False) - parse_rules( - rules_file=self.rules_file, - rule_type='retro', - diameters=diam, - outfile=outfile.name - ) - # Test if outfile has one single line (header) - self.assertEqual(len(outfile.readlines()), 1) - outfile.close() - unlink(outfile.name) + output = parse_rules( + None, rules_file=self.rules_file, rule_type="retro", diameters="3" + ) + self.assertEqual(len(output.splitlines()), 1) def test_OneDiameter(self): - for diam in ['2']: - with self.subTest(diam=diam): - outfile = NamedTemporaryFile(delete=False) - parse_rules( - rules_file=self.rules_file, - diameters=diam, - outfile=outfile.name - ) - self.assertListEqual( - list(io_open(outfile.name)), - list(io_open(self.ref_d2_csv)) - ) - outfile.close() - unlink(outfile.name) + # A closed path can be reopened by pandas on Windows. + with TemporaryDirectory() as directory: + outfile = Path(directory) / "rules.csv" + parse_rules(str(outfile), rules_file=self.rules_file, diameters="2") + self.assertEqual(outfile.read_text(), Path(self.ref_d2_csv).read_text()) def test_MiscDiametersArgument(self): - for diam in ['2-']: - with self.subTest(diam=diam): - outfile = NamedTemporaryFile(delete=True) - self.assertRaises( - ValueError, - parse_rules, - rules_file='retrorules', - diameters=diam, - outfile=outfile.name - ) + self.assertRaises(ValueError, parse_rules, None, diameters="2-") diff --git a/tests/test_downloads.py b/tests/test_downloads.py new file mode 100644 index 0000000..f04c42d --- /dev/null +++ b/tests/test_downloads.py @@ -0,0 +1,282 @@ +import gzip +import tarfile +from io import BytesIO, StringIO +from pathlib import Path +from tempfile import TemporaryDirectory +from unittest.mock import patch +from urllib.error import URLError + +import pandas as pd +import pytest + +from rrparser import build_args_parser, list_retrorules, parse_rules +from rrparser.Args import DEFAULT_RULES_DIR +from rrparser.Downloads import RR02_CACHE_FILE, RR02_FILE, RR02_URL, fetch_dataset + +PAGE = b"""CSV +CSV +JSON +Legacy""" +CSV = b"""TEMPLATE_ID,SCORE,TEMPLATE,REACTIONS,RADII,ECS +RR:1,0.8,"[C,N]>>[O]",RHEA:1,1;4,1.1.1.1;2.1.1 +RR:2,0.5,[O]>>[C],RHEA:2,10, +""" + + +def test_discovery_download_and_offline_cache(): + with TemporaryDirectory() as directory: + with patch("rrparser.Downloads.urlopen", return_value=BytesIO(PAGE)): + releases = list_retrorules() + assert list(releases) == [ + ("3.1.0", "rhea"), + ("3.0", "rhea"), + ("rr02", "metanetx"), + ] + assert releases[("rr02", "metanetx")] == RR02_URL + with patch( + "rrparser.Downloads.urlopen", + side_effect=[BytesIO(PAGE), BytesIO(gzip.compress(CSV))], + ): + filename = fetch_dataset(directory, release="latest", dataset="rhea") + assert gzip.decompress(Path(filename).read_bytes()) == CSV + with patch( + "rrparser.Downloads.urlopen", side_effect=AssertionError("Network used") + ): + assert ( + fetch_dataset(directory, release="latest", dataset="rhea") == filename + ) + assert fetch_dataset(directory, "3.1.0", "rhea") == filename + with patch( + "rrparser.Downloads.urlopen", return_value=BytesIO(PAGE) + ), pytest.raises(ValueError, match="unavailable"): + fetch_dataset(directory, "9.0", "rhea") + with patch( + "rrparser.Downloads.urlopen", return_value=BytesIO(PAGE) + ), pytest.raises(ValueError, match="unavailable"): + fetch_dataset(directory, release="latest", dataset="metanetx") + + +def test_failed_download_does_not_pollute_cache(): + with TemporaryDirectory() as directory: + with ( + patch( + "rrparser.Downloads.urlopen", + side_effect=[BytesIO(PAGE), URLError("offline")], + ), + pytest.raises(ValueError, match="Cannot download"), + ): + fetch_dataset(directory, release="latest", dataset="rhea") + assert not list(Path(directory).iterdir()) + + +@pytest.mark.parametrize("release", ["9.0", "rr03", "3.x", ""]) +def test_wrong_release_shows_available_choices(release): + with TemporaryDirectory() as directory: + with patch( + "rrparser.Downloads.urlopen", return_value=BytesIO(PAGE) + ) as request, pytest.raises(ValueError) as error: + fetch_dataset(directory, release, "rhea") + message = str(error.value) + assert repr(release) in message or f"{release} / rhea" in message + assert "Available releases and datasets:" in message + assert "3.1.0\trhea" in message + assert "3.0\trhea" in message + assert "rr02\tmetanetx" in message + request.assert_called_once() + assert not list(Path(directory).iterdir()) + + +def test_wrong_release_offline_preserves_error(): + with TemporaryDirectory() as directory: + with patch( + "rrparser.Downloads.urlopen", side_effect=URLError("offline") + ), pytest.raises(ValueError) as error: + fetch_dataset(directory, "rr03") + assert "Invalid release 'rr03'" in str(error.value) + assert "Available releases could not be retrieved" in str(error.value) + + +def test_rr02_download_cache_and_parsing(): + rules = (Path(__file__).parent / "data/rules.csv").read_bytes() + archive = BytesIO() + with tarfile.open(fileobj=archive, mode="w:gz") as tar: + for name, data in [(RR02_FILE, rules), ("../../unwanted.txt", b"unwanted")]: + member = tarfile.TarInfo(name) + member.size = len(data) + tar.addfile(member, BytesIO(data)) + original_archive = archive.getvalue() + archive.seek(0) + with TemporaryDirectory() as directory: + with patch("rrparser.Downloads.urlopen", return_value=archive) as download: + filename = fetch_dataset(directory, "rr02") + download.assert_called_once_with(RR02_URL, timeout=30) + assert filename == str(Path(directory) / RR02_CACHE_FILE) + assert not (Path(directory) / RR02_FILE).parent.exists() + assert Path(filename).read_bytes() == original_archive + with patch( + "rrparser.Downloads.urlopen", side_effect=AssertionError("Network used") + ): + output = parse_rules( + None, + rules_dir=directory, + release="rr02", + rule_type="retro", + diameters="2", + ) + parsed = pd.read_csv(StringIO(output)) + assert not parsed.empty + assert set(parsed["Rule usage"]) <= {"retro", "both"} + assert set(parsed["Diameter"]) == {2} + assert list(Path(directory).rglob("*.txt")) == [] + assert list(Path(directory).rglob("*.csv")) == [] + assert Path(filename).read_bytes() == original_archive + with pytest.raises(ValueError, match="only available for metanetx"): + fetch_dataset(directory, "rr02", "rhea") + + +def test_rr02_failed_download_leaves_no_cache(): + with TemporaryDirectory() as directory: + with patch( + "rrparser.Downloads.urlopen", side_effect=URLError("offline") + ), pytest.raises(ValueError, match="Cannot download"): + fetch_dataset(directory, "rr02") + assert not list(Path(directory).iterdir()) + + +def test_rr02_missing_csv(): + archive = BytesIO() + with tarfile.open(fileobj=archive, mode="w:gz"): + pass + archive.seek(0) + with TemporaryDirectory() as directory: + with patch("rrparser.Downloads.urlopen", return_value=archive), pytest.raises( + ValueError, match="does not contain" + ): + parse_rules(None, rules_dir=directory, release="rr02") + assert not list(Path(directory).rglob("*.csv")) + assert not list(Path(directory).rglob("*.part")) + + +def test_templates_filters_scores_and_output(): + with TemporaryDirectory() as directory: + rules = Path(directory) / "rules.csv.gz" + rules.write_bytes(gzip.compress(CSV)) + scores = Path(directory) / "scores.csv" + scores.write_text("Rule ID,Score\nRR:1,0.9\n") + output = parse_rules( + None, + rules_file=str(rules), + diameters="8", + ec=["1.1"], + scores_file=str(scores), + output_format="tsv", + ) + parsed = pd.read_csv(StringIO(output), sep="\t") + assert parsed["Rule ID"].tolist() == ["RR:1"] + assert parsed["Score"].tolist() == [0.9] + assert parsed["Rule"].tolist() == ["[C,N]>>[O]"] + assert ( + len( + pd.read_csv( + StringIO(parse_rules(None, rules_file=str(rules), diameters="20")) + ) + ) + == 1 + ) + with pytest.raises(ValueError, match="no retro/forward"): + parse_rules(None, rules_file=str(rules), rule_type="retro") + + +def test_default_release_ignores_latest_cache(): + args = build_args_parser().parse_args([]) + assert args.release == "3.1.0" + assert args.diameters == ",".join(str(diameter) for diameter in range(0, 21, 2)) + with TemporaryDirectory() as directory: + cache = Path(directory) + rules = cache / "retrorules-3.1.0-metanetx.csv.gz" + rules.write_bytes(gzip.compress(CSV)) + (cache / "retrorules-latest-metanetx.txt").write_text("3.0") + (cache / "retrorules-3.0-metanetx.csv.gz").write_bytes(b"old cache") + with patch( + "rrparser.Downloads.urlopen", side_effect=AssertionError("Network used") + ): + assert fetch_dataset(directory) == str(rules) + output = parse_rules(None, rules_dir=directory) + assert pd.read_csv(StringIO(output))["Rule ID"].tolist() == ["RR:1", "RR:2"] + + +def test_all_template_diameters(): + with TemporaryDirectory() as directory: + rules = Path(directory) / "templates.csv" + pd.DataFrame( + { + "TEMPLATE_ID": [f"RR:{radius}" for radius in range(11)], + "TEMPLATE": ["[C]>>[O]"] * 11, + "SCORE": [1.0] * 11, + "ECS": ["1.1.1.1"] * 11, + "RADII": list(range(11)), + } + ).to_csv(rules, index=False) + output = parse_rules(None, rules_file=str(rules)) + assert len(pd.read_csv(StringIO(output))) == 11 + for diameter in range(0, 21, 2): + output = parse_rules(None, rules_file=str(rules), diameters=str(diameter)) + assert pd.read_csv(StringIO(output))["Rule ID"].tolist() == [ + f"RR:{diameter // 2}" + ] + output = parse_rules(None, rules_file=str(rules), diameters="0,18,20") + assert pd.read_csv(StringIO(output))["Rule ID"].tolist() == [ + "RR:0", + "RR:9", + "RR:10", + ] + for invalid in ["1", "3", "19", "21", "22", "-2"]: + with pytest.raises(ValueError): + parse_rules(None, rules_file=str(rules), diameters=invalid) + + +def test_cache_and_filtered_output_are_separate(): + args = build_args_parser().parse_args( + ["--dataset", "rhea", "--radii", "4", "-o", "rules.csv"] + ) + assert args.rules_dir == DEFAULT_RULES_DIR + assert ( + Path(DEFAULT_RULES_DIR) + == Path(__file__).resolve().parents[1] / "rrparser" / "cache" + ) + assert ( + build_args_parser().parse_args(["--cache-dir", "./shared-cache"]).rules_dir + == "./shared-cache" + ) + with pytest.raises(SystemExit): + build_args_parser().parse_args(["--radii", "4", "--diameters", "8"]) + with TemporaryDirectory() as directory: + cache = Path(directory) / "cache" + output = Path(directory) / "rules.csv" + with patch( + "rrparser.Downloads.urlopen", + side_effect=[BytesIO(PAGE), BytesIO(gzip.compress(CSV))], + ): + assert ( + parse_rules( + str(output), + rules_dir=str(cache), + dataset=args.dataset, + radii=args.radii, + ) + is None + ) + source = cache / "retrorules-3.1.0-rhea.csv.gz" + assert gzip.decompress(source.read_bytes()) == CSV + assert pd.read_csv(output)["Rule ID"].tolist() == ["RR:1"] + with patch( + "rrparser.Downloads.urlopen", side_effect=AssertionError("Network used") + ): + parse_rules(str(output), rules_dir=str(cache), dataset="rhea", radii="10") + assert pd.read_csv(output)["Rule ID"].tolist() == ["RR:2"] + assert gzip.decompress(source.read_bytes()) == CSV + assert not (cache / "rules.csv").exists() + assert not (Path(directory) / source.name).exists() + for invalid in ["-1", "11", "1.5", ""]: + with pytest.raises(ValueError, match="Radii must"): + parse_rules(None, rules_file=str(source), radii=invalid) diff --git a/tests/test_filters.py b/tests/test_filters.py index b64d563..e0b24ed 100644 --- a/tests/test_filters.py +++ b/tests/test_filters.py @@ -5,70 +5,62 @@ """ # Generic for test process +from os import path + +# Specific for tests themselves +from pandas import read_csv from test_main import Test_RR # Specific for tool from rrparser.Parser import filter_, read_ecnumbers -# Specific for tests themselves -from pandas import read_csv, DataFrame -from os import path - - # Cette classe est un groupe de tests pour la fonction filter_. Son nom DOIT commencer # par 'Test' et la classe DOIT hériter de unittest.TestCase. # 'Test_' prefix is mandatory class Test_RR_Filters(Test_RR): - def setUp(self): super().setUp() self.here = path.abspath(path.dirname(__file__)) - self.rules_file = path.join(self.here, 'data', 'rules.csv') + self.rules_file = path.join(self.here, "data", "rules.csv") self.df = read_csv(self.rules_file) def test_filter_by_rule_type(self): - for rule_type in ['forward', 'retro', 'all']: + for rule_type in ["forward", "retro", "all"]: with self.subTest(rule_type=rule_type): df_expected = read_csv( - path.join(self.here, 'data', f'rules_{rule_type}.csv') + path.join(self.here, "data", f"rules_{rule_type}.csv") ) df = filter_( df=self.df, rule_type=rule_type, - diameters=[2,4,6,8,10,12,14,16], - ecx=[] + diameters=[2, 4, 6, 8, 10, 12, 14, 16], + ecx=[], ) # reindex df df = df.reset_index(drop=True) self.assertTrue(df.equals(df_expected)) def test_filter_by_diameters(self): - for diameters in [[2,4,6,8,10,12,14,16], [2]]: + for diameters in [[2, 4, 6, 8, 10, 12, 14, 16], [2]]: with self.subTest(diameters=diameters): df_expected = read_csv( - path.join(self.here, 'data', f'rules_d_{"-".join(str(x) for x in diameters)}.csv') - ) - df = filter_( - df=self.df, - rule_type='all', - diameters=diameters, - ecx=[] + path.join( + self.here, + "data", + f"rules_d_{'-'.join(str(x) for x in diameters)}.csv", + ) ) + df = filter_(df=self.df, rule_type="all", diameters=diameters, ecx=[]) # reindex df df = df.reset_index(drop=True) self.assertTrue(df.equals(df_expected)) - + def test_filter_by_ecx(self): - ecx = read_ecnumbers(path.join(self.here, 'data', 'ec.csv')) - df_expected = read_csv( - path.join(self.here, 'data', 'rules_ecx.csv') - ) + ecx = read_ecnumbers(path.join(self.here, "data", "ec.csv")) + df_expected = read_csv(path.join(self.here, "data", "rules_ecx.csv")) df = filter_( - df=self.df, - rule_type='all', - diameters=[2,4,6,8,10,12,14,16], - ecx=ecx + df=self.df, rule_type="all", diameters=[2, 4, 6, 8, 10, 12, 14, 16], ecx=ecx ) # reindex df df = df.reset_index(drop=True) diff --git a/tests/test_input-format.py b/tests/test_input-format.py index 33eac9d..e108741 100644 --- a/tests/test_input-format.py +++ b/tests/test_input-format.py @@ -1,75 +1,25 @@ -""" -Created on June 17 2020 - -@author: Joan Hérisson -""" - -# Generic for test process from test_main import Test_RR -# Specific for tool from rrparser import parse_rules -# Specific for tests themselves -from hashlib import sha256 -from pathlib import Path -from tempfile import NamedTemporaryFile - - -# Cette classe est un groupe de tests. Son nom DOIT commencer -# par 'Test' et la classe DOIT hériter de unittest.TestCase. -# 'Test_' prefix is mandatory class Test_RR_InputFormat(Test_RR): - - - # def test_GoodInputFormatCSV(self): - # diam = '2' - # outfile = NamedTemporaryFile(delete=True) - # parse_rules(rules_file='data/rules.csv', - # input_format='csv', - # rule_type='retro', - # diameters=diam, - # outfile=outfile.name) - # self.assertEqual( - # sha256(Path(outfile.name).read_bytes()).hexdigest(), self.hash_d2_csv) - # outfile.close() - - def test_BadInputFormatCSV_1(self): - diam = '2' - outfile = NamedTemporaryFile(delete=True) self.assertRaises( KeyError, parse_rules, - rules_file=self.rules_file, - input_format='tsv', - diameters=diam, - outfile=outfile.name + None, + rules_file=self.rules_file, + input_format="tsv", + diameters="2", ) - outfile.close() - def test_BadInputFormatCSV_2(self): - diam = '2' - outfile = NamedTemporaryFile(delete=True) - self.assertRaises(ValueError, - parse_rules, - rules_file=self.rules_file, - input_format='other', - diameters=diam, - outfile=outfile.name) - outfile.close() - - - # def test_SmallRulesFile_OneDiameter_WithFingerPrint(self): - # for diam in ['2']: - # with self.subTest(diam=diam): - # tempdir = mkdtemp(suffix='_'+diam) - # outfile = self.rr_parser.parse_rules(outdir=tempdir, - # rules_file='data/rules.csv', - # diameters=diam) - # self.assertEqual( - # sha256(Path(outfile).read_bytes()).hexdigest(), - # 'a6c2852a991e394bdbaf04791a90e803d4410a53f037165a7f08956edde63066' - # ) + self.assertRaises( + ValueError, + parse_rules, + None, + rules_file=self.rules_file, + input_format="other", + diameters="2", + ) diff --git a/tests/test_light.py b/tests/test_light.py index 7451752..a3fe904 100644 --- a/tests/test_light.py +++ b/tests/test_light.py @@ -1,88 +1,40 @@ -""" -Created on June 17 2020 +from pathlib import Path +from tempfile import TemporaryDirectory -@author: Joan Hérisson -""" -from os import unlink - -# Generic for test process from test_main import Test_RR -# Specific for tool -from rrparser import ( - parse_rules, - __path__ as pkg_path -) - -# Specific for tests themselves -from io import open as io_open -from tempfile import NamedTemporaryFile +from rrparser import parse_rules -# Cette classe est un groupe de tests. Son nom DOIT commencer -# par 'Test' et la classe DOIT hériter de unittest.TestCase. -# 'Test_' prefix is mandatory class Test_RR_Light(Test_RR): - - - # def test_SmallRulesFile_OneDiameter(self): - # for diam in ['2,4,6']: - # with self.subTest(diam=diam): - # outfile = NamedTemporaryFile(delete=True) - # parse_rules( - # rules_file='data/rules.csv', - # diameters=diam, - # outfile=outfile.name - # ) - # self.assertEqual( - # sha256( - # Path(outfile.name).read_bytes() - # ).hexdigest(), - # self.hash_d2_csv - # ) - # outfile.close() - - def test_GoodInputFormatCSV(self): - diam = '2' - outfile = NamedTemporaryFile(delete=False) - parse_rules( - rules_file = self.rules_file, - input_format = 'csv', - diameters = diam, - outfile = outfile.name - ) - self.assertListEqual( - list(io_open(outfile.name)), - list(io_open(self.ref_d2_csv)) - ) - outfile.close() - unlink(outfile.name) - + # Do not keep an output file handle open while pandas writes it. + with TemporaryDirectory() as directory: + outfile = Path(directory) / "rules.csv" + parse_rules( + str(outfile), + rules_file=self.rules_file, + input_format="csv", + diameters="2", + ) + self.assertEqual(outfile.read_text(), Path(self.ref_d2_csv).read_text()) def test_BadInputFormatCSV_1(self): - diam = '2' - outfile = NamedTemporaryFile(suffix='_'+diam, delete=True) self.assertRaises( KeyError, parse_rules, - rules_file = self.rules_file, - input_format = 'tsv', - diameters = diam, - outfile = outfile.name + None, + rules_file=self.rules_file, + input_format="tsv", + diameters="2", ) - outfile.close() - def test_BadInputFormatCSV_2(self): - diam = '2' - outfile = NamedTemporaryFile(delete=True) self.assertRaises( ValueError, parse_rules, - rules_file = self.rules_file, - input_format = 'other', - diameters = diam, - outfile = outfile.name + None, + rules_file=self.rules_file, + input_format="other", + diameters="2", ) - outfile.close() diff --git a/tests/test_main.py b/tests/test_main.py index ce3f9d9..8773772 100644 --- a/tests/test_main.py +++ b/tests/test_main.py @@ -5,8 +5,9 @@ """ # Generic for test process -from unittest import TestCase from os import path as os_path +from unittest import TestCase + # Specific for tool from rrparser import __path__ as pkg_path @@ -15,23 +16,9 @@ # par 'Test' et la classe DOIT hériter de unittest.TestCase. # 'Test_' prefix is mandatory class Test_RR(TestCase): - - - diameters = ['2', '4', '6', '8', '10', '12', '14', '16'] - data = os_path.join( - os_path.dirname(os_path.realpath(__file__)), - 'data' - ) - rules_file = os_path.join( - data, - 'rules.csv' - ) - ref_d2_csv = os_path.join( - data, - 'out_d2.csv' - ) - ref_d2_tsv = os_path.join( - data, - 'out_d2.tsv' - ) - pkg_path = pkg_path + diameters = ("2", "4", "6", "8", "10", "12", "14", "16") + data = os_path.join(os_path.dirname(os_path.realpath(__file__)), "data") + rules_file = os_path.join(data, "rules.csv") + ref_d2_csv = os_path.join(data, "out_d2.csv") + ref_d2_tsv = os_path.join(data, "out_d2.tsv") + pkg_path = pkg_path diff --git a/tests/test_misc.py b/tests/test_misc.py index 3c88e13..38335f4 100644 --- a/tests/test_misc.py +++ b/tests/test_misc.py @@ -5,92 +5,64 @@ """ # Generic for test process -from test_main import Test_RR # Specific for tool -from sys import path as sys_path -from os import path as os_path -from os import unlink -from rrparser import parse_rules # Specific for tests themselves -from itertools import combinations -from random import sample, seed -from io import open as io_open -from tempfile import NamedTemporaryFile +from pathlib import Path +from tarfile import open as open_tar +from tempfile import TemporaryDirectory +from unittest.mock import patch + +from pandas import read_csv +from test_main import Test_RR + +from rrparser import parse_rules +from rrparser.Downloads import RR02_CACHE_FILE, RR02_FILE # Cette classe est un groupe de tests. Son nom DOIT commencer # par 'Test' et la classe DOIT hériter de unittest.TestCase. # 'Test_' prefix is mandatory class Test_Misc(Test_RR): - - # # 'test_' prefix is mandatory - # def test_Precedence(self): - # diam = '2' - # tempdir = TemporaryDirectory(suffix='_'+diam) - # outfile = self.rr_parser.parse_rules(outdir=tempdir.name, - # rules_file='data/rules.csv', - # rule_type='retro', - # diameters=diam) - # self.assertEqual( - # sha256(Path(outfile).read_bytes()).hexdigest(), self.hash_d2_csv) - def test_SmallRulesFile_OneDiameter_SpecifyOutfile(self): - for format in ['csv', 'tsv']: - with self.subTest(format=format): - diam = '2' - outfile = NamedTemporaryFile(delete=False) - parse_rules( - rules_file = self.rules_file, - outfile = outfile.name, - diameters = diam, - output_format = format - ) - self.assertListEqual( - list( - io_open( - outfile.name - ) - ), - list( - io_open( - getattr( - self, - 'ref_d2_'+format - ) - ) - ) - ) - outfile.close() - unlink(outfile.name) - - - def test_AllTypes_RandomDiam(self): - for rule_type in ['all', 'retro', 'forward']: - # for i in range(len(self.diameters)): - i = 3 - diams = list( - combinations( - self.diameters, - i+1 - ) - ) - seed(2) - sub_diams = sample(diams, 1) - for diam in sub_diams: - with self.subTest(rule_type=rule_type, diam=diam): - outfile = NamedTemporaryFile(delete=False) + with TemporaryDirectory() as directory: + for format in ["csv", "tsv"]: + with self.subTest(format=format): + outfile = Path(directory) / ("rules." + format) parse_rules( - rules_file = 'retrorules', - outfile = outfile.name, - rule_type = rule_type, - diameters = ','.join(diam) + str(outfile), + rules_file=self.rules_file, + diameters="2", + output_format=format, ) - # Test if outfile has more than one single line (header) - self.assertGreater( - len(outfile.readlines()), - 1 - ) - outfile.close() - unlink(outfile.name) + reference = Path(getattr(self, "ref_d2_" + format)) + self.assertEqual(outfile.read_text(), reference.read_text()) + + def test_AllTypes_CachedRR02(self): + diameters = [2, 4, 8, 14] + with TemporaryDirectory() as directory: + cache = Path(directory) / "cache" + cache.mkdir() + with open_tar(cache / RR02_CACHE_FILE, mode="w:gz") as archive: + archive.add(self.rules_file, arcname=RR02_FILE) + outfile = Path(directory) / "rules.csv" + with patch( + "rrparser.Downloads.urlopen", side_effect=AssertionError("Network used") + ): + for rule_type in ["all", "retro", "forward"]: + with self.subTest(rule_type=rule_type): + parse_rules( + outfile=str(outfile), + rules_dir=str(cache), + release="rr02", + rule_type=rule_type, + diameters=",".join(map(str, diameters)), + ) + results = read_csv(outfile) + self.assertFalse(results.empty) + self.assertTrue(set(results["Diameter"]).issubset(diameters)) + if rule_type != "all": + self.assertTrue( + set(results["Rule usage"]).issubset({"both", rule_type}) + ) diff --git a/tests/test_rule-type.py b/tests/test_rule-type.py index b819fb8..4c9aca8 100644 --- a/tests/test_rule-type.py +++ b/tests/test_rule-type.py @@ -1,46 +1,16 @@ -""" -Created on June 17 2020 - -@author: Joan Hérisson -""" - -# Generic for test process from test_main import Test_RR -# Specific for tool from rrparser.Parser import parse_rules -# Specific for tests themselves -from hashlib import sha256 -from pathlib import Path -from tempfile import NamedTemporaryFile - -# Cette classe est un groupe de tests. Son nom DOIT commencer -# par 'Test' et la classe DOIT hériter de unittest.TestCase. -# 'Test_' prefix is mandatory class Test_RR_RuleType(Test_RR): - def test_BadRuleTypeArgument(self): - for rule_type in ['test', 'reto']: + for rule_type in ["test", "reto"]: with self.subTest(rule_type=rule_type): - outfile = NamedTemporaryFile(delete=True) self.assertRaises( ValueError, parse_rules, - rules_file = 'retrorules', - outfile = outfile.name, - rule_type = rule_type, - diameters = '2' + None, + rule_type=rule_type, + diameters="2", ) - - # def test_EmptyRuleTypeArgument(self): - # for rule_type in ['']: - # with self.subTest(rule_type=rule_type): - # outfile = NamedTemporaryFile(delete=True) - # self.assertRaises(ValueError, - # parse_rules, - # rules_file=fetch_retrorules(), - # outfile=outfile.name, - # rule_type=rule_type, - # diameters='2') diff --git a/tests/test_scores-repl.py b/tests/test_scores-repl.py index 1e9e8a5..4f0ac30 100644 --- a/tests/test_scores-repl.py +++ b/tests/test_scores-repl.py @@ -5,46 +5,38 @@ """ # Generic for test process +from os import path + +# Specific for tests themselves +from pandas import read_csv from test_main import Test_RR # Specific for tool from rrparser.Parser import replace_scores -# Specific for tests themselves -from pandas import read_csv, DataFrame -from os import path - - # Cette classe est un groupe de tests pour la fonction filter_. Son nom DOIT commencer # par 'Test' et la classe DOIT hériter de unittest.TestCase. # 'Test_' prefix is mandatory class Test_RR_ScoreReplacing(Test_RR): - def setUp(self): super().setUp() self.here = path.abspath(path.dirname(__file__)) - self.rules_file = path.join(self.here, 'data', 'rules_forward.csv') + self.rules_file = path.join(self.here, "data", "rules_forward.csv") self.df = read_csv(self.rules_file) - def test_score_replacing(self): # Read rules df = read_csv(self.rules_file) # Read rules scores - rules_scores_file = path.join(self.here, 'data', 'scores.csv') + rules_scores_file = path.join(self.here, "data", "scores.csv") df_scores = read_csv(rules_scores_file) # Replace scores - df_replaced = replace_scores( - results=df, - scores=df_scores - ) + df_replaced = replace_scores(results=df, scores=df_scores) # Read expected results - df_expected = read_csv( - path.join(self.here, 'data', 'rules_scores.csv') - ) + df_expected = read_csv(path.join(self.here, "data", "rules_scores.csv")) # reindex df df_replaced = df_replaced.reset_index(drop=True) df_expected = df_expected.reset_index(drop=True) # Compare dataframes - self.assertTrue(df_replaced.equals(df_expected)) \ No newline at end of file + self.assertTrue(df_replaced.equals(df_expected))