cmd cleanup, step context builder removed

carnival/__init__.py

@@ -6,7 +6,6 @@
 from carnival import cmd
 from carnival import internal_tasks
 from carnival.utils import log
-from carnival.context import context_ref
 
 
 if not sys.warnoptions:
@@ -23,6 +22,5 @@
     'TaskBase', 'StepsTask',
     'cmd',
     'log',
-    'context_ref',
     'internal_tasks',
 ]

carnival/cli.py

@@ -1,6 +1,7 @@
 import os
 import sys
-from typing import Any, Dict, Iterable, Type
+import typing
+import collections
 
 import click
 import dotenv
@@ -23,10 +24,10 @@ def is_completion_script(complete_var: str) -> bool:
     return os.getenv(complete_var, None) is not None
 
 
-task_types: Dict[str, Type[TaskBase]] = {}
+task_types: typing.OrderedDict[str, typing.Type[TaskBase]] = collections.OrderedDict()
 
 
-def except_hook(type: Type[Any], value: Any, traceback: Any) -> None:
+def except_hook(type: typing.Type[typing.Any], value: typing.Any, traceback: typing.Any) -> None:
     print(f"{type.__name__}: {value} \nYou can use --debug flag to see full traceback.")
 
 
@@ -49,7 +50,7 @@ def main() -> int:
     @click.command()
     @click.option('--debug', is_flag=True, default=False, help="Turn on debug mode")
     @click.argument('tasks', required=True, type=click.Choice(list(task_types.keys())), nargs=-1)
-    def cli(debug: bool, tasks: Iterable[str]) -> None:
+    def cli(debug: bool, tasks: typing.Iterable[str]) -> None:
         if debug is True:
             print("Debug mode on.")
         else:

carnival/cmd/__init__.py

@@ -7,14 +7,12 @@
 Основные шаги доступны в отдельном репозитории: <https://github.com/carnival-org/carnival-contrib>.
 """
 
-from carnival.cmd import cli, system, systemd, apt, transfer, fs
+from carnival.cmd import cli, system, transfer, fs
 
 
 __all__ = [
     'cli',
     'system',
-    'systemd',
-    'apt',
     'transfer',
     'fs',
 ]

carnival/cmd/cli.py

@@ -1,25 +1,11 @@
-from typing import Any
-
-from carnival import connection
+from carnival.host import AnyConnection
 from invoke import Result  # type: ignore
 
 
-def _run_command(command: str, **kwargs: Any) -> Result:
-    assert connection.conn is not None, "No connection"
-    return connection.conn.run(command, **kwargs)
-
-
-def run(command: str, **kwargs: Any) -> Result:
+def run(c: AnyConnection, command: str, warn: bool = True, hide: bool = False) -> Result:
     """
     Запустить комманду
-    """
-    return _run_command(command, **kwargs)
-
-
-def pty(command: str, **kwargs: Any) -> Result:
-    """
-    Запустить комманду, используя псевдотерминальную сессию
 
     См <https://docs.pyinvoke.org/en/latest/api/runners.html>
     """
-    return run(command, pty=True, **kwargs)
+    return c.run(command, pty=True, warn=warn, hide=hide)

carnival/cmd/fs.py

@@ -1,32 +1,45 @@
 from typing import List, Optional
+import re
+
+from carnival import cmd
+from carnival.host import AnyConnection
 
-from carnival import cmd, connection
 from invoke import Result  # type: ignore
-from patchwork import files  # type:ignore
 
 
-def mkdirs(*dirs: str) -> List[Result]:
+def _escape_for_regex(text: str) -> str:
+    """Escape ``text`` to allow literal matching using egrep"""
+    regex = re.escape(text)
+    # Seems like double escaping is needed for \
+    regex = regex.replace("\\\\", "\\\\\\")
+    # Triple-escaping seems to be required for $ signs
+    regex = regex.replace(r"\$", r"\\\$")
+    # Whereas single quotes should not be escaped
+    regex = regex.replace(r"\'", "'")
+    return regex
+
+
+def mkdirs(c: AnyConnection, *dirs: str) -> List[Result]:
     """
     Создать директории
 
     :param dirs: пути которые нужно создать
     """
-    return [cmd.cli.run(f"mkdir -p {x}", hide=True) for x in dirs]
+    return [cmd.cli.run(c, f"mkdir -p {x}", hide=True) for x in dirs]
 
 
-def is_dir_exists(dir_path: str) -> bool:
+def is_dir_exists(c: AnyConnection, dir_path: str) -> bool:
     """
     Узнать существует ли директория
 
     :param dir_path: путь до директории
     """
-    return bool(cmd.cli.run(f"test -d {dir_path}", warn=True, hide=True).ok)
+    return bool(cmd.cli.run(c, f"test -d {dir_path}", warn=True, hide=True).ok)
 
 
-def is_file_contains(filename: str, text: str, exact: bool = False, escape: bool = True) -> bool:
+def is_file_contains(c: AnyConnection, filename: str, text: str, exact: bool = False, escape: bool = True) -> bool:
     """
     Содержит ли файл текст
-    См <https://fabric-patchwork.readthedocs.io/en/latest/api/files.html#patchwork.files.contains>
 
     :param filename: путь до файла
     :param text: текст который нужно искать
@@ -35,26 +48,27 @@ def is_file_contains(filename: str, text: str, exact: bool = False, escape: bool
 
 
     """
-    assert connection.conn is not None, "No connection"
-    return bool(files.contains(
-        connection.conn,
-        runner=connection.conn.run,
-        filename=filename, text=text, exact=exact, escape=escape
-    ))
+    if escape:
+        text = _escape_for_regex(text)
+        if exact:
+            text = "^{}$".format(text)
+    egrep_cmd = 'egrep "{}" "{}"'.format(text, filename)
+    return c.run(egrep_cmd, hide=True, warn=True).ok  # type: ignore
 
 
-def is_file_exists(path: str) -> bool:
+def is_file_exists(c: AnyConnection, path: str) -> bool:
     """
     Проверить существует ли файл
-    <https://fabric-patchwork.readthedocs.io/en/latest/api/files.html#patchwork.files.exists>
 
     :param path: путь до файла
     """
-    assert connection.conn is not None, "No connection"
-    return bool(files.exists(connection.conn, runner=connection.conn.run, path=path))
+
+    cmd = 'test -e "$(echo {})"'.format(path)
+    return c.run(cmd, hide=True, warn=True).ok  # type: ignore
 
 
 def ensure_dir_exists(
+    c: AnyConnection,
     path: str,
     user: Optional[str] = None,
     group: Optional[str] = None,
@@ -63,12 +77,15 @@ def ensure_dir_exists(
     """
     Проверить что директория существует и параметры соответствуют заданным
 
-    <https://fabric-patchwork.readthedocs.io/en/latest/api/files.html#patchwork.files.directory>
-
     :param path: путь до директории
     :param user: владелец
     :param group: группа
     :param mode: права
     """
-    assert connection.conn is not None, "No connection"
-    files.directory(connection.conn, runner=connection.conn.run, path=path, user=user, group=group, mode=mode)
+
+    c.run("mkdir -p {}".format(path))
+    if user is not None:
+        group = group or user
+        c.run("chown {}:{} {}".format(user, group, path))
+    if mode is not None:
+        c.run("chmod {} {}".format(mode, path))