-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Organize package and start testing (#15)
* Organize package Split commands into their own modules for better management. - Add new init command - Add new edit command - Add new implode command - Add new sniff command - Add some unit tests - Update validations/confirmations * Reformat file
- Loading branch information
Showing
15 changed files
with
517 additions
and
155 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -1,164 +1,34 @@ | ||
import configparser | ||
import sys | ||
from pathlib import Path | ||
from typing import NoReturn, Union | ||
|
||
import click | ||
|
||
from repo_man.commands.add import add | ||
from repo_man.commands.edit import edit | ||
from repo_man.commands.flavors import flavors | ||
from repo_man.commands.implode import implode | ||
from repo_man.commands.init import init | ||
from repo_man.commands.list_repos import list_repos | ||
from repo_man.commands.sniff import sniff | ||
from repo_man.consts import REPO_TYPES_CFG | ||
|
||
FAIL = "\033[91m" | ||
ENDC = "\033[0m" | ||
REPO_TYPES_CFG = "repo-man.cfg" | ||
|
||
|
||
def check_if_allowed(path: Path) -> Union[bool, NoReturn]: | ||
if REPO_TYPES_CFG not in (str(item) for item in path.iterdir()): | ||
print(f"{FAIL}The current directory is not configured for repository management{ENDC}") | ||
sys.exit(1) | ||
|
||
return True | ||
|
||
|
||
def parse_repo_types(config: configparser.ConfigParser) -> dict[str, set[str]]: | ||
repo_types: dict[str, set[str]] = {"all": set()} | ||
for section in config.sections(): | ||
repos = {repo for repo in config[section]["known"].split("\n") if repo} | ||
repo_types[section] = repos | ||
if section != "ignore": | ||
repo_types["all"].update(repos) | ||
|
||
return repo_types | ||
|
||
|
||
def check_missing_repos(path: Path, repo_types: dict[str, set[str]]) -> None: | ||
missing = set() | ||
directories = {str(directory) for directory in path.iterdir()} | ||
|
||
for repo in sorted(repo_types["all"]): | ||
if repo not in directories: | ||
missing.add(repo) | ||
|
||
if missing: | ||
print(f"{FAIL}The following repositories are configured but do not exist:") | ||
for repo in missing: | ||
print(f"\t{repo}") | ||
sys.exit(1) | ||
|
||
return None | ||
|
||
|
||
def get_valid_repo_types(): | ||
config = configparser.ConfigParser() | ||
config.read(REPO_TYPES_CFG) | ||
valid_repo_types = parse_repo_types(config) | ||
return sorted(set(valid_repo_types.keys())) | ||
|
||
|
||
def main(): | ||
path = Path(".") | ||
check_if_allowed(path) | ||
|
||
config = configparser.ConfigParser() | ||
config.read(REPO_TYPES_CFG) | ||
valid_repo_types = parse_repo_types(config) | ||
check_missing_repos(path, valid_repo_types) | ||
|
||
cli() | ||
|
||
|
||
@click.group(invoke_without_command=True, context_settings={"help_option_names": ["-h", "--help"]}) | ||
@click.group(context_settings={"help_option_names": ["-h", "--help"]}) | ||
@click.version_option(package_name="repo-man") | ||
@click.option("-k", "--known", is_flag=True, help="List known repository types") | ||
@click.option("-u", "--unconfigured", is_flag=True, help="List repositories without a configured type") | ||
@click.option("-d", "--duplicates", is_flag=True, help="List repositories with more than one configured type") | ||
# @click.option("-v", "--verbose", "verbosity", count=True) | ||
def cli(known: bool, unconfigured: bool, duplicates: bool): | ||
@click.pass_context | ||
def cli(context): | ||
"""Manage repositories of different types""" | ||
|
||
path = Path(".") | ||
config = configparser.ConfigParser() | ||
config.read(REPO_TYPES_CFG) | ||
valid_repo_types = parse_repo_types(config) | ||
|
||
if known: | ||
known_repo_types = sorted( | ||
[repo_type for repo_type in valid_repo_types if repo_type != "all" and repo_type != "ignore"] | ||
) | ||
for repo_type in known_repo_types: | ||
print(repo_type) | ||
|
||
if unconfigured: | ||
for directory in sorted(path.iterdir()): | ||
if ( | ||
directory.is_dir() | ||
and str(directory) not in valid_repo_types["all"] | ||
and str(directory) not in valid_repo_types.get("ignore", []) | ||
): | ||
print(directory) | ||
|
||
if duplicates: | ||
seen = set() | ||
for repo_type in valid_repo_types: | ||
if repo_type != "all" and repo_type != "ignore": | ||
for repo in valid_repo_types[repo_type]: | ||
if repo in seen: | ||
print(repo) | ||
seen.add(repo) | ||
|
||
|
||
@cli.command(name="list", help="The type of repository to manage") | ||
@click.option("-t", "--type", "repo_type", type=click.Choice(get_valid_repo_types()), show_choices=False, required=True) | ||
def list_repos(repo_type: str): | ||
"""List matching repositories""" | ||
|
||
config = configparser.ConfigParser() | ||
config.read(REPO_TYPES_CFG) | ||
valid_repo_types = parse_repo_types(config) | ||
context.obj = config | ||
|
||
for repo in valid_repo_types[repo_type]: | ||
print(repo) | ||
|
||
return None | ||
|
||
|
||
@cli.command | ||
@click.argument("repo", type=click.Path(exists=True, file_okay=False)) | ||
def flavors(repo: str): | ||
"""List the configured types for a repository""" | ||
|
||
config = configparser.ConfigParser() | ||
config.read(REPO_TYPES_CFG) | ||
|
||
found = set() | ||
|
||
for section in config.sections(): | ||
if section == "ignore": | ||
continue | ||
if repo in config[section]["known"].split("\n"): | ||
found.add(section) | ||
|
||
for repository in sorted(found): | ||
print(repository) | ||
|
||
|
||
@cli.command | ||
@click.option("-t", "--type", "repo_types", multiple=True, help="The type of the repository", required=True) | ||
@click.argument("repo", type=click.Path(exists=True, file_okay=False)) | ||
def add(repo: str, repo_types: list): | ||
"""Add a new repository""" | ||
|
||
config = configparser.ConfigParser() | ||
config.read(REPO_TYPES_CFG) | ||
|
||
for repo_type in repo_types: | ||
if repo_type in config: | ||
original_config = config[repo_type]["known"] | ||
else: | ||
original_config = "" | ||
config.add_section(repo_type) | ||
|
||
if "known" not in config[repo_type] or repo not in config[repo_type]["known"].split("\n"): | ||
config.set(repo_type, "known", f"{original_config}\n{repo}") | ||
|
||
with open(REPO_TYPES_CFG, "w") as config_file: | ||
config.write(config_file) | ||
def main(): | ||
cli.add_command(add) | ||
cli.add_command(edit) | ||
cli.add_command(flavors) | ||
cli.add_command(implode) | ||
cli.add_command(init) | ||
cli.add_command(list_repos) | ||
cli.add_command(sniff) | ||
cli() |
Empty file.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,36 @@ | ||
import configparser | ||
from pathlib import Path | ||
|
||
import click | ||
|
||
from repo_man.consts import REPO_TYPES_CFG | ||
from repo_man.utils import pass_config | ||
|
||
|
||
@click.command | ||
@click.option("-t", "--type", "repo_types", multiple=True, help="The type of the repository", required=True) | ||
@click.argument("repo", type=click.Path(exists=True, file_okay=False)) | ||
@pass_config | ||
def add(config: configparser.ConfigParser, repo: str, repo_types: list[str]): | ||
"""Add a new repository""" | ||
|
||
if not Path(REPO_TYPES_CFG).exists(): | ||
click.confirm(click.style(f"No {REPO_TYPES_CFG} file found. Do you want to continue?", fg="yellow"), abort=True) | ||
|
||
new_types = [repo_type for repo_type in repo_types if repo_type not in config] | ||
if new_types: | ||
message = "\n\t".join(new_types) | ||
click.confirm(f"The following types are unknown and will be added:\n\n\t{message}\n\nContinue?", abort=True) | ||
|
||
for repo_type in repo_types: | ||
if repo_type in config: | ||
original_config = config[repo_type]["known"] | ||
else: | ||
original_config = "" | ||
config.add_section(repo_type) | ||
|
||
if "known" not in config[repo_type] or repo not in config[repo_type]["known"].split("\n"): | ||
config.set(repo_type, "known", f"{original_config}\n{repo}") | ||
|
||
with open(REPO_TYPES_CFG, "w") as config_file: | ||
config.write(config_file) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,16 @@ | ||
from pathlib import Path | ||
|
||
import click | ||
|
||
from repo_man.consts import REPO_TYPES_CFG | ||
|
||
|
||
@click.command | ||
def edit(): | ||
"""Edit the repo-man configuration manually""" | ||
|
||
if not Path(REPO_TYPES_CFG).exists(): | ||
click.echo(click.style(f"No {REPO_TYPES_CFG} file found.", fg="red")) | ||
return | ||
|
||
click.edit(filename=REPO_TYPES_CFG) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,29 @@ | ||
import configparser | ||
from pathlib import Path | ||
|
||
import click | ||
|
||
from repo_man.consts import REPO_TYPES_CFG | ||
from repo_man.utils import pass_config | ||
|
||
|
||
@click.command | ||
@click.argument("repo", type=click.Path(exists=True, file_okay=False)) | ||
@pass_config | ||
def flavors(config: configparser.ConfigParser, repo: str): | ||
"""List the configured types for a repository""" | ||
|
||
if not Path(REPO_TYPES_CFG).exists(): | ||
click.echo(click.style(f"No {REPO_TYPES_CFG} file found.", fg="red")) | ||
return | ||
|
||
found = set() | ||
|
||
for section in config.sections(): | ||
if section == "ignore": | ||
continue | ||
if repo in config[section]["known"].split("\n"): | ||
found.add(section) | ||
|
||
for repository in sorted(found): | ||
click.echo(repository) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,14 @@ | ||
from pathlib import Path | ||
|
||
import click | ||
|
||
from repo_man.consts import REPO_TYPES_CFG | ||
|
||
|
||
@click.command | ||
@click.argument("path", type=click.Path(exists=True, file_okay=False, dir_okay=True, path_type=Path)) | ||
def implode(path: Path): | ||
"""Remove repo-man configuration for the specified directory""" | ||
|
||
click.confirm(click.style("Are you sure you want to do this?", fg="yellow"), abort=True) | ||
(path / REPO_TYPES_CFG).unlink(missing_ok=True) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,20 @@ | ||
from pathlib import Path | ||
|
||
import click | ||
|
||
from repo_man.consts import REPO_TYPES_CFG | ||
|
||
|
||
@click.command | ||
@click.argument("path", type=click.Path(exists=True, file_okay=False, dir_okay=True, path_type=Path)) | ||
def init(path: Path): | ||
"""Initialize repo-man to track repositories located at the specified path""" | ||
|
||
if (path / REPO_TYPES_CFG).exists(): | ||
click.confirm( | ||
click.style(f"{REPO_TYPES_CFG} file already exists. Overwrite with empty configuration?", fg="yellow"), | ||
abort=True, | ||
) | ||
|
||
with open(path / REPO_TYPES_CFG, "w"): | ||
pass |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,27 @@ | ||
import configparser | ||
from pathlib import Path | ||
|
||
import click | ||
|
||
from repo_man.consts import REPO_TYPES_CFG | ||
from repo_man.utils import get_valid_repo_types, parse_repo_types, pass_config | ||
|
||
|
||
@click.command(name="list", help="The type of repository to manage") | ||
@click.option("-t", "--type", "repo_type", type=click.Choice(get_valid_repo_types()), show_choices=False, required=True) | ||
@pass_config | ||
def list_repos(config: configparser.ConfigParser, repo_type: str): | ||
"""List matching repositories""" | ||
|
||
if not Path(REPO_TYPES_CFG).exists(): | ||
click.echo(click.style(f"No {REPO_TYPES_CFG} file found.", fg="red")) | ||
return | ||
|
||
valid_repo_types = parse_repo_types(config) | ||
|
||
repos = sorted(valid_repo_types[repo_type]) | ||
|
||
if len(repos) > 25: | ||
click.echo_via_pager("\n".join(repos)) | ||
else: | ||
click.echo("\n".join(repos)) |
Oops, something went wrong.