CopyBot/Red-DiscordBot
1
0
Fork 0
mirror of https://github.com/Cog-Creators/Red-DiscordBot.git synced 2025-11-06 11:18:54 -05:00
Code Issues Packages Projects Releases Wiki Activity

Added Downloader cog (#786)

Browse Source
This commit is contained in:
Will 2017-06-17 19:31:32 -04:00 committed by Twentysix
parent b12a41cd77
commit 53810b2262
16 changed files with 1461 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
.gitignore vendored
View File

@ -6,3 +6,4 @@ __pycache__
*.log
*.tmp
.data
lib/

6
cogs/downloader/__init__.py Normal file
View File

@ -0,0 +1,6 @@
from core.bot import Red
from .downloader import Downloader
def setup(bot: Red):
bot.add_cog(Downloader(bot))

45
cogs/downloader/checks.py Normal file
View File

@ -0,0 +1,45 @@
import asyncio
import discord
from discord.ext import commands
__all__ = ["install_agreement", ]
REPO_INSTALL_MSG = (
"You're about to add a 3rd party repository. The creator of Red"
" and its community have no responsibility for any potential "
"damage that the content of 3rd party repositories might cause."
"\n\nBy typing '**I agree**' you declare that you have read and"
" fully understand the above message. This message won't be "
"shown again until the next reboot.\n\nYou have **30** seconds"
" to reply to this message."
)
def install_agreement():
async def pred(ctx: commands.Context):
downloader = ctx.command.instance
if downloader is None:
return True
elif downloader.already_agreed:
return True
elif ctx.invoked_subcommand is None or \
isinstance(ctx.invoked_subcommand, commands.Group):
return True
def does_agree(msg: discord.Message):
return ctx.author == msg.author and \
ctx.channel == msg.channel and \
msg.content == "I agree"
await ctx.send(REPO_INSTALL_MSG)
try:
await ctx.bot.wait_for('message', check=does_agree, timeout=30)
except asyncio.TimeoutError:
await ctx.send("Your response has timed out, please try again.")
return False
downloader.already_agreed = True
return True
return commands.check(pred)

24
cogs/downloader/converters.py Normal file
View File

@ -0,0 +1,24 @@
import discord
from discord.ext import commands
from .repo_manager import RepoManager
from .installable import Installable
class RepoName(commands.Converter):
async def convert(self, ctx: commands.Context, arg: str) -> str:
return RepoManager.validate_and_normalize_repo_name(arg)
class InstalledCog(commands.Converter):
async def convert(self, ctx: commands.Context, arg: str) -> Installable:
downloader = ctx.bot.get_cog("Downloader")
if downloader is None:
raise commands.CommandError("Downloader not loaded.")
cog = discord.utils.get(downloader.installed_cogs, name=arg)
if cog is None:
raise commands.BadArgument(
"That cog is not installed"
)
return cog

330
cogs/downloader/downloader.py Normal file
View File

@ -0,0 +1,330 @@
import os
import shutil
from typing import Tuple
import discord
from discord.ext import commands
from pathlib import Path
from sys import path as syspath
from core import Config
from core.bot import Red
from core import checks
from core.utils.chat_formatting import box
from .repo_manager import RepoManager, Repo
from .installable import Installable
from .converters import RepoName, InstalledCog
from .log import log
from .errors import CloningError, ExistingGitRepo
from .checks import install_agreement
class Downloader:
COG_PATH = Path.cwd() / "cogs"
LIB_PATH = Path.cwd() / "lib"
SHAREDLIB_PATH = LIB_PATH / "cog_shared"
SHAREDLIB_INIT = SHAREDLIB_PATH / "__init__.py"
def __init__(self, bot: Red):
self.bot = bot
self.conf = Config.get_conf(self, unique_identifier=998240343,
force_registration=True)
self.conf.register_global(
repos={},
installed=[]
)
self.already_agreed = False
self.LIB_PATH.mkdir(parents=True, exist_ok=True)
self.SHAREDLIB_PATH.mkdir(parents=True, exist_ok=True)
if not self.SHAREDLIB_INIT.exists():
with self.SHAREDLIB_INIT.open(mode='w') as _:
pass
if str(self.LIB_PATH) not in syspath:
syspath.insert(1, str(self.LIB_PATH))
self._repo_manager = RepoManager(self.conf)
@property
def installed_cogs(self) -> Tuple[Installable]:
"""
Returns the dictionary mapping cog name to install location
and repo name.
:return:
"""
installed = self.conf.installed()
# noinspection PyTypeChecker
return tuple(Installable.from_json(v) for v in installed)
async def _add_to_installed(self, cog: Installable):
"""
Marks a cog as installed.
:param cog:
:return:
"""
installed = self.conf.installed()
cog_json = cog.to_json()
if cog_json not in installed:
installed.append(cog_json)
await self.conf.set("installed", installed)
async def _remove_from_installed(self, cog: Installable):
"""
Removes a cog from the saved list of installed cogs.
:param cog:
:return:
"""
installed = self.conf.installed()
cog_json = cog.to_json()
if cog_json in installed:
installed.remove(cog_json)
await self.conf.set("installed", installed)
async def _reinstall_cogs(self, cogs: Tuple[Installable]) -> Tuple[Installable]:
"""
Installs a list of cogs, used when updating.
:param cogs:
:return: Any cogs that failed to copy
"""
failed = []
for cog in cogs:
if not await cog.copy_to(self.COG_PATH):
failed.append(cog)
# noinspection PyTypeChecker
return tuple(failed)
async def _reinstall_libraries(self, cogs: Tuple[Installable]) -> Tuple[Installable]:
"""
Reinstalls any shared libraries from the repos of cogs that
were updated.
:param cogs:
:return: Any libraries that failed to copy
"""
repo_names = set(cog.repo_name for cog in cogs)
unfiltered_repos = (self._repo_manager.get_repo(r) for r in repo_names)
repos = filter(lambda r: r is not None, unfiltered_repos)
failed = []
for repo in repos:
if not await repo.install_libraries(target_dir=self.SHAREDLIB_PATH):
failed.extend(repo.available_libraries)
# noinspection PyTypeChecker
return tuple(failed)
async def _reinstall_requirements(self, cogs: Tuple[Installable]) -> bool:
"""
Reinstalls requirements for given cogs that have been updated.
Returns a bool that indicates if all requirement installations
were successful.
:param cogs:
:return:
"""
# Reduces requirements to a single list with no repeats
requirements = set(r for c in cogs for r in c.requirements)
repo_names = self._repo_manager.get_all_repo_names()
repos = [(self._repo_manager.get_repo(rn), []) for rn in repo_names]
# This for loop distributes the requirements across all repos
# which will allow us to concurrently install requirements
for i, req in enumerate(requirements):
repo_index = i % len(repos)
repos[repo_index][1].append(req)
has_reqs = list(filter(lambda item: len(item[1]) > 0, repos))
ret = True
for repo, reqs in has_reqs:
for req in reqs:
# noinspection PyTypeChecker
ret = ret and await repo.install_raw_requirements([req, ], self.LIB_PATH)
return ret
@staticmethod
async def _delete_cog(target: Path):
"""
Removes an (installed) cog.
:param target: Path pointing to an existing file or directory
:return:
"""
if not target.exists():
return
if target.is_dir():
shutil.rmtree(str(target))
elif target.is_file():
os.remove(str(target))
@commands.group()
@checks.is_owner()
async def repo(self, ctx):
"""
Command group for managing Downloader repos.
"""
if ctx.invoked_subcommand is None:
await self.bot.send_cmd_help(ctx)
@repo.command(name="add")
@install_agreement()
async def _repo_add(self, ctx, name: RepoName, repo_url: str, branch: str=None):
"""
Add a new repo to Downloader.
:param name: Name that must follow python variable naming rules and
contain only characters A-Z and numbers 0-9 and _
:param repo_url: Clone url for the cog repo
:param branch: Specify branch if you want it to be different than
repo default.
"""
try:
# noinspection PyTypeChecker
await self._repo_manager.add_repo(
name=name,
url=repo_url,
branch=branch
)
except ExistingGitRepo:
await ctx.send("That git repo has already been added under another name.")
except CloningError:
await ctx.send("Something went wrong during the cloning process.")
log.exception("Something went wrong during the cloning process.")
else:
await ctx.send("Repo `{}` successfully added.".format(name))
@repo.command(name="del")
async def _repo_del(self, ctx, repo_name: Repo):
"""
Removes a repo from Downloader and its' files.
:param repo_name: Repo name in Downloader
"""
await self._repo_manager.delete_repo(repo_name.name)
await ctx.send("The repo `{}` has been deleted successfully.".format(repo_name.name))
@repo.command(name="list")
async def _repo_list(self, ctx):
"""
Lists all installed repos.
"""
repos = self._repo_manager.get_all_repo_names()
joined = "Installed Repos:\n" + "\n".join(["+ " + r for r in repos])
await ctx.send(box(joined, lang="diff"))
@commands.group()
@checks.is_owner()
async def cog(self, ctx):
"""
Command group for managing installable Cogs.
"""
if ctx.invoked_subcommand is None:
await self.bot.send_cmd_help(ctx)
@cog.command(name="install")
async def _cog_install(self, ctx, repo_name: Repo, cog_name: str):
"""
Installs a cog from the given repo.
:param repo_name:
:param cog_name: Cog name available from `[p]cog list <repo_name>`
"""
cog = discord.utils.get(repo_name.available_cogs, name=cog_name)
if cog is None:
await ctx.send("Error, there is no cog by the name of"
" `{}` in the `{}` repo.".format(cog_name, repo_name.name))
return
if not await repo_name.install_requirements(cog, self.LIB_PATH):
await ctx.send("Failed to install the required libraries for"
" `{}`: `{}`".format(cog.name, cog.requirements))
return
await repo_name.install_cog(cog, self.COG_PATH)
await self._add_to_installed(cog)
await repo_name.install_libraries(self.SHAREDLIB_PATH)
await ctx.send("`{}` cog successfully installed.".format(cog_name))
@cog.command(name="uninstall")
async def _cog_uninstall(self, ctx, cog_name: InstalledCog):
"""
Allows you to uninstall cogs that were previously installed
through Downloader.
:param cog_name:
"""
# noinspection PyUnresolvedReferences,PyProtectedMember
real_name = cog_name.name
poss_installed_path = self.COG_PATH / real_name
if poss_installed_path.exists():
await self._delete_cog(poss_installed_path)
# noinspection PyTypeChecker
await self._remove_from_installed(cog_name)
await ctx.send("`{}` was successfully removed.".format(real_name))
else:
await ctx.send("That cog was installed but can no longer"
" be located. You may need to remove it's"
" files manually if it is still usable.")
@cog.command(name="update")
async def _cog_update(self, ctx, cog_name: InstalledCog=None):
"""
Updates all cogs or one of your choosing.
:param cog_name:
"""
if cog_name is None:
updated = await self._repo_manager.update_all_repos()
installed_cogs = set(self.installed_cogs)
updated_cogs = set(cog for repo in updated.keys() for cog in repo.available_cogs)
installed_and_updated = updated_cogs & installed_cogs
# noinspection PyTypeChecker
await self._reinstall_requirements(installed_and_updated)
# noinspection PyTypeChecker
await self._reinstall_cogs(installed_and_updated)
# noinspection PyTypeChecker
await self._reinstall_libraries(installed_and_updated)
await ctx.send("Cog update completed successfully.")
@cog.command(name="list")
async def _cog_list(self, ctx, repo_name: Repo):
"""
Lists all available cogs from a single repo.
:param repo_name: Repo name available from `[p]repo list`
"""
cogs = repo_name.available_cogs
cogs = "Available Cogs:\n" + "\n".join(
["+ {}: {}".format(c.name, c.short or "") for c in cogs])
await ctx.send(box(cogs, lang="diff"))
@cog.command(name="info")
async def _cog_info(self, ctx, repo_name: Repo, cog_name: str):
"""
Lists information about a single cog.
:param repo_name:
:param cog_name:
"""
cog = discord.utils.get(repo_name.available_cogs, name=cog_name)
if cog is None:
await ctx.send("There is no cog `{}` in the repo `{}`".format(
cog_name, repo_name.name
))
return
msg = "Information on {}:\n{}".format(cog.name, cog.description or "")
await ctx.send(box(msg))

79
cogs/downloader/errors.py Normal file
View File

@ -0,0 +1,79 @@
class DownloaderException(Exception):
"""
Base class for Downloader exceptions.
"""
pass
class GitException(DownloaderException):
"""
Generic class for git exceptions.
"""
class InvalidRepoName(DownloaderException):
"""
Throw when a repo name is invalid. Check
the message for a more detailed reason.
"""
pass
class ExistingGitRepo(DownloaderException):
"""
Thrown when trying to clone into a folder where a
git repo already exists.
"""
pass
class MissingGitRepo(DownloaderException):
"""
Thrown when a git repo is expected to exist but
does not.
"""
pass
class CloningError(GitException):
"""
Thrown when git clone returns a non zero exit code.
"""
pass
class CurrentHashError(GitException):
"""
Thrown when git returns a non zero exit code attempting
to determine the current commit hash.
"""
pass
class HardResetError(GitException):
"""
Thrown when there is an issue trying to execute a hard reset
(usually prior to a repo update).
"""
pass
class UpdateError(GitException):
"""
Thrown when git pull returns a non zero error code.
"""
pass
class GitDiffError(GitException):
"""
Thrown when a git diff fails.
"""
pass
class PipError(DownloaderException):
"""
Thrown when pip returns a non-zero return code.
"""
pass

190
cogs/downloader/installable.py Normal file
View File

@ -0,0 +1,190 @@
import json
import distutils.dir_util
import shutil
from enum import Enum
from pathlib import Path
from typing import Union, MutableMapping, Any
from .log import log
from .json_mixins import RepoJSONMixin
class InstallableType(Enum):
UNKNOWN = 0
COG = 1
SHARED_LIBRARY = 2
class Installable(RepoJSONMixin):
"""
Base class for anything the Downloader cog can install.
- Modules
- Repo Libraries
- Other stuff?
"""
INFO_FILE_DESCRIPTION = """
The info.json file may exist inside every package folder in the repo,
it is optional however. This string describes the valid keys within
an info file (and maybe how the Downloader cog uses them).
KEYS (case sensitive):
author (list of strings) - list of names of authors of the cog
bot_version (list of integer) - Min version number of Red in the
format (MAJOR, MINOR, PATCH)
description (string) - A long description of the cog that appears
when a user executes `!cog info`
hidden (bool) - Determines if a cog is available for install.
install_msg (string) - The message that gets displayed when a cog is
installed
required_cogs (map of cogname to repo URL) - A map of required cogs
that this cog depends on. Downloader will not deal with this
functionality but it may be useful for other cogs.
requirements (list of strings) - list of required libraries that are
passed to pip on cog install. SHARED_LIBRARIES do NOT go in this
list.
short (string) - A short description of the cog that appears when
a user executes `!cog list`
tags (list of strings) - A list of strings that are related to the
functionality of the cog. Used to aid in searching.
type (string) - Optional, defaults to COG. Must be either COG or
SHARED_LIBRARY. If SHARED_LIBRARY then HIDDEN will be True.
"""
def __init__(self, location: Path):
"""
Base installable initializer.
:param location: Location (file or folder) to the installable.
"""
super().__init__(location)
self._location = location
self.repo_name = self._location.parent.stem
self.author = ()
self.bot_version = (3, 0, 0)
self.hidden = False
self.required_cogs = {} # Cog name -> repo URL
self.requirements = ()
self.tags = ()
self.type = InstallableType.UNKNOWN
if self._info_file.exists():
self._process_info_file(self._info_file)
if self._info == {}:
self.type = InstallableType.COG
def __eq__(self, other):
# noinspection PyProtectedMember
return self._location == other._location
def __hash__(self):
return hash(self._location)
@property
def name(self):
return self._location.stem
async def copy_to(self, target_dir: Path) -> bool:
"""
Copies this cog/shared_lib to the given directory. This
will overwrite any files in the target directory
:param target_dir: The installation directory to install to.
:return: bool - status of installation
"""
if self._location.is_file():
copy_func = shutil.copy2
else:
copy_func = distutils.dir_util.copy_tree
# noinspection PyBroadException
try:
copy_func(
src=str(self._location),
dst=str(target_dir / self._location.stem)
)
except:
log.exception("Error occurred when copying path:"
" {}".format(self._location))
return False
return True
def _read_info_file(self):
super()._read_info_file()
if self._info_file.exists():
self._process_info_file()
def _process_info_file(self, info_file_path: Path=None) -> MutableMapping[str, Any]:
"""
Processes an information file. Loads dependencies among other
information into this object.
:type info_file_path:
:param info_file_path: Optional path to information file, defaults to `self.__info_file`
:return: Raw information dictionary
"""
info_file_path = info_file_path or self._info_file
if info_file_path is None or not info_file_path.is_file():
raise ValueError("No valid information file path was found.")
info = {}
with info_file_path.open(encoding='utf-8') as f:
try:
info = json.load(f)
except json.JSONDecodeError:
info = {}
log.exception("Invalid JSON information file at path:"
" {}".format(info_file_path))
else:
self._info = info
try:
author = tuple(info.get("author", []))
except ValueError:
author = ()
self.author = author
try:
bot_version = tuple(info.get("bot_version", [3, 0, 0]))
except ValueError:
bot_version = 2
self.bot_version = bot_version
try:
hidden = bool(info.get("hidden", False))
except ValueError:
hidden = False
self.hidden = hidden
self.required_cogs = info.get("required_cogs", {})
self.requirements = info.get("requirements", ())
try:
tags = tuple(info.get("tags", ()))
except ValueError:
tags = ()
self.tags = tags
installable_type = info.get("type", "")
if installable_type in ("", "COG"):
self.type = InstallableType.COG
elif installable_type == "SHARED_LIBRARY":
self.type = InstallableType.SHARED_LIBRARY
self.hidden = True
else:
self.type = InstallableType.UNKNOWN
return info
def to_json(self):
return {
"location": self._location.relative_to(Path.cwd()).parts
}
@classmethod
def from_json(cls, data: dict):
location = Path.cwd() / Path(*data["location"])
return cls(location=location)

37
cogs/downloader/json_mixins.py Normal file
View File

@ -0,0 +1,37 @@
import json
from pathlib import Path
class RepoJSONMixin:
INFO_FILE_NAME = "info.json"
def __init__(self, repo_folder: Path):
self._repo_folder = repo_folder
self.author = None
self.install_msg = None
self.short = None
self.description = None
self._info_file = repo_folder / self.INFO_FILE_NAME
if self._info_file.exists():
self._read_info_file()
self._info = {}
def _read_info_file(self):
if not (self._info_file.exists() or self._info_file.is_file()):
return
try:
with self._info_file.open(encoding='utf-8') as f:
info = json.load(f)
except json.JSONDecodeError:
return
else:
self._info = info
self.author = info.get("author")
self.install_msg = info.get("install_msg")
self.short = info.get("short")
self.description = info.get("description")

3
cogs/downloader/log.py Normal file
View File

@ -0,0 +1,3 @@
import logging
log = logging.getLogger("red.downloader")

529
cogs/downloader/repo_manager.py Normal file
View File

@ -0,0 +1,529 @@
import asyncio
import json
import os
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path
from typing import Tuple, MutableMapping
from subprocess import run as sp_run, PIPE
from sys import executable
import pkgutil
import shutil
import functools
from discord.ext import commands
from core import Config
from .errors import *
from .installable import Installable, InstallableType
from .log import log
from .json_mixins import RepoJSONMixin
class Repo(RepoJSONMixin):
GIT_CLONE = "git clone -b {branch} {url} {folder}"
GIT_CLONE_NO_BRANCH = "git clone {url} {folder}"
GIT_CURRENT_BRANCH = "git -C {path} rev-parse --abbrev-ref HEAD"
GIT_LATEST_COMMIT = "git -C {path} rev-parse {branch}"
GIT_HARD_RESET = "git -C {path} reset --hard origin/{branch} -q"
GIT_PULL = "git -C {path} pull -q --ff-only"
GIT_DIFF_FILE_STATUS = ("git -C {path} diff --no-commit-id --name-status"
" {old_hash} {new_hash}")
GIT_LOG = ("git -C {path} log --relative-date --reverse {old_hash}.."
" {relative_file_path}")
PIP_INSTALL = "{python} -m pip install -U -t {target_dir} {reqs}"
def __init__(self, name: str, url: str, branch: str, folder_path: Path,
available_modules: Tuple[Installable]=(), loop: asyncio.AbstractEventLoop=None):
self.url = url
self.branch = branch
self.name = name
self.folder_path = folder_path
self.folder_path.mkdir(parents=True, exist_ok=True)
super().__init__(self.folder_path)
self.available_modules = available_modules
self._executor = ThreadPoolExecutor(1)
self._repo_lock = asyncio.Lock()
self._loop = loop
if self._loop is None:
self._loop = asyncio.get_event_loop()
@classmethod
async def convert(cls, ctx: commands.Context, argument: str):
downloader_cog = ctx.bot.get_cog("Downloader")
if downloader_cog is None:
raise commands.CommandError("No Downloader cog found.")
# noinspection PyProtectedMember
repo_manager = downloader_cog._repo_manager
poss_repo = repo_manager.get_repo(argument)
if poss_repo is None:
raise commands.BadArgument("Repo by the name {} does not exist.".format(argument))
return poss_repo
def _existing_git_repo(self) -> (bool, Path):
git_path = self.folder_path / '.git'
return git_path.exists(), git_path
async def _get_file_update_statuses(
self, old_hash: str, new_hash: str) -> MutableMapping[str, str]:
"""
Gets the file update status letters for each changed file between
the two hashes.
:param old_hash: Pre-update
:param new_hash: Post-update
:return: Mapping of filename -> status_letter
"""
p = await self._run(
self.GIT_DIFF_FILE_STATUS.format(
path=self.folder_path,
old_hash=old_hash,
new_hash=new_hash
)
)
if p.returncode != 0:
raise GitDiffError("Git diff failed for repo at path:"
" {}".format(self.folder_path))
stdout = p.stdout.strip().decode().split('\n')
ret = {}
for filename in stdout:
# TODO: filter these filenames by ones in self.available_modules
status, _, filepath = filename.partition('\t')
ret[filepath] = status
return ret
async def _get_commit_notes(self, old_commit_hash: str,
relative_file_path: str) -> str:
"""
Gets the commit notes from git log.
:param old_commit_hash: Point in time to start getting messages
:param relative_file_path: Path relative to the repo folder of the file
to get messages for.
:return: Git commit note log
"""
p = await self._run(
self.GIT_LOG.format(
path=self.folder_path,
old_hash=old_commit_hash,
relative_file_path=relative_file_path
)
)
if p.returncode != 0:
raise GitException("An exception occurred while executing git log on"
" this repo: {}".format(self.folder_path))
return p.stdout.decode().strip()
def _update_available_modules(self) -> Tuple[str]:
"""
Updates the available modules attribute for this repo.
:return: List of available modules.
"""
curr_modules = []
"""
for name in self.folder_path.iterdir():
if name.is_dir():
spec = importlib.util.spec_from_file_location(
name.stem, location=str(name.parent)
)
if spec is not None:
curr_modules.append(
Installable(location=name)
)
"""
for file_finder, name, is_pkg in pkgutil.walk_packages(path=[str(self.folder_path), ]):
curr_modules.append(
Installable(location=self.folder_path / name)
)
self.available_modules = curr_modules
# noinspection PyTypeChecker
return tuple(self.available_modules)
async def _run(self, *args, **kwargs):
env = os.environ.copy()
env['GIT_TERMINAL_PROMPT'] = '0'
kwargs['env'] = env
async with self._repo_lock:
return await self._loop.run_in_executor(
self._executor,
functools.partial(sp_run, *args, stdout=PIPE, **kwargs)
)
async def clone(self) -> Tuple[str]:
"""
Clones a new repo.
:return: List of available modules from this repo.
"""
exists, path = self._existing_git_repo()
if exists:
raise ExistingGitRepo(
"A git repo already exists at path: {}".format(path)
)
if self.branch is not None:
p = await self._run(
self.GIT_CLONE.format(
branch=self.branch,
url=self.url,
folder=self.folder_path
).split()
)
else:
p = await self._run(
self.GIT_CLONE_NO_BRANCH.format(
url=self.url,
folder=self.folder_path
).split()
)
if p.returncode != 0:
raise CloningError("Error when running git clone.")
if self.branch is None:
self.branch = await self.current_branch()
self._read_info_file()
return self._update_available_modules()
async def current_branch(self) -> str:
"""
Determines the current branch using git commands.
:return: Current branch name
"""
exists, _ = self._existing_git_repo()
if not exists:
raise MissingGitRepo(
"A git repo does not exist at path: {}".format(self.folder_path)
)
p = await self._run(
self.GIT_CURRENT_BRANCH.format(
path=self.folder_path
).split()
)
if p.returncode != 0:
raise GitException("Could not determine current branch"
" at path: {}".format(self.folder_path))
return p.stdout.decode().strip()
async def current_commit(self, branch: str=None) -> str:
"""
Determines the current commit hash of the repo.
:param branch: Override for repo's branch attribute
:return: Commit hash string
"""
if branch is None:
branch = self.branch
exists, _ = self._existing_git_repo()
if not exists:
raise MissingGitRepo(
"A git repo does not exist at path: {}".format(self.folder_path)
)
p = await self._run(
self.GIT_LATEST_COMMIT.format(
path=self.folder_path,
branch=branch
).split()
)
if p.returncode != 0:
raise CurrentHashError("Unable to determine old commit hash.")
return p.stdout.decode().strip()
async def hard_reset(self, branch: str=None) -> None:
"""
Performs a hard reset on the current repo.
:param branch: Override for repo branch attribute.
:return:
"""
if branch is None:
branch = self.branch
exists, _ = self._existing_git_repo()
if not exists:
raise MissingGitRepo(
"A git repo does not exist at path: {}".format(self.folder_path)
)
p = await self._run(
self.GIT_HARD_RESET.format(
path=self.folder_path,
branch=branch
).split()
)
if p.returncode != 0:
raise HardResetError("Some error occurred when trying to"
" execute a hard reset on the repo at"
" the following path: {}".format(self.folder_path))
async def update(self) -> (str, str):
"""
Updates the current branch of this repo.
:return: Old commit hash
:return: New commit hash
"""
curr_branch = await self.current_branch()
old_commit = await self.current_commit(branch=curr_branch)
await self.hard_reset(branch=curr_branch)
p = await self._run(
self.GIT_PULL.format(
path=self.folder_path
).split()
)
if p.returncode != 0:
raise UpdateError("Git pull returned a non zero exit code"
" for the repo located at path: {}".format(self.folder_path))
new_commit = await self.current_commit(branch=curr_branch)
self._update_available_modules()
self._read_info_file()
return old_commit, new_commit
async def install_cog(self, cog: Installable, target_dir: Path) -> bool:
"""
Copies a cog to the target directory.
:param cog:
:param target_dir:
:return: bool - if installation succeeded
"""
if cog not in self.available_cogs:
raise DownloaderException("That cog does not exist in this repo")
if not target_dir.is_dir():
raise ValueError("That target directory is not actually a directory.")
if not target_dir.exists():
raise ValueError("That target directory does not exist.")
return await cog.copy_to(target_dir=target_dir)
async def install_libraries(self, target_dir: Path, libraries: Tuple[Installable]=()) -> bool:
"""
Copies all shared libraries (or a given subset) to the target
directory.
:param target_dir:
:param libraries: A subset of available libraries
:return: bool - all copies succeeded
"""
if libraries:
if not all([i in self.available_libraries for i in libraries]):
raise ValueError("Some given libraries are not available in this repo.")
else:
libraries = self.available_libraries
if libraries:
return all([lib.copy_to(target_dir=target_dir) for lib in libraries])
return True
async def install_requirements(self, cog: Installable, target_dir: Path) -> bool:
"""
Installs the requirements defined by the requirements
attribute on the cog object and puts them in the given
target directory.
:param cog:
:param target_dir:
:return:
"""
if not target_dir.is_dir():
raise ValueError("Target directory is not a directory.")
target_dir.mkdir(parents=True, exist_ok=True)
return await self.install_raw_requirements(cog.requirements, target_dir)
async def install_raw_requirements(self, requirements: Tuple[str], target_dir: Path) -> bool:
"""
Installs a list of requirements using pip and places them into
the given target directory.
:param requirements:
:param target_dir:
:return:
"""
if len(requirements) == 0:
return True
# TODO: Check and see if any of these modules are already available
p = await self._run(
self.PIP_INSTALL.format(
python=executable,
target_dir=target_dir,
reqs=" ".join(requirements)
).split()
)
if p.returncode != 0:
log.error("Something went wrong when installing"
" the following requirements:"
" {}".format(", ".join(requirements)))
return False
return True
@property
def available_cogs(self) -> Tuple[Installable]:
"""
Returns a list of available cogs (not shared libraries and not hidden).
:return: tuple(installable)
"""
# noinspection PyTypeChecker
return tuple(
[m for m in self.available_modules
if m.type == InstallableType.COG and not m.hidden]
)
@property
def available_libraries(self) -> Tuple[Installable]:
"""
Returns a list of available shared libraries in this repo.
"""
# noinspection PyTypeChecker
return tuple(
[m for m in self.available_modules
if m.type == InstallableType.SHARED_LIBRARY]
)
def to_json(self):
return {
"url": self.url,
"name": self.name,
"branch": self.branch,
"folder_path": self.folder_path.relative_to(Path.cwd()).parts,
"available_modules": [m.to_json() for m in self.available_modules]
}
@classmethod
def from_json(cls, data):
# noinspection PyTypeChecker
return Repo(data['name'], data['url'], data['branch'],
Path.cwd() / Path(*data['folder_path']),
tuple([Installable.from_json(m) for m in data['available_modules']]))
class RepoManager:
def __init__(self, downloader_config: Config):
self.downloader_config = downloader_config
self.repos_folder = Path(__file__).parent / 'repos'
self._repos = self._load_repos() # str_name: Repo
def does_repo_exist(self, name: str) -> bool:
return name in self._repos
@staticmethod
def validate_and_normalize_repo_name(name: str) -> str:
if not name.isidentifier():
raise InvalidRepoName("Not a valid Python variable name.")
return name.lower()
async def add_repo(self, url: str, name: str, branch: str="master") -> Repo:
"""
Adds a repo and clones it.
:param url:
:param name:
:param branch:
:return:
"""
name = self.validate_and_normalize_repo_name(name)
if self.does_repo_exist(name):
raise InvalidRepoName(
"That repo name you provided already exists."
" Please choose another."
)
# noinspection PyTypeChecker
r = Repo(url=url, name=name, branch=branch,
folder_path=self.repos_folder / name)
await r.clone()
self._repos[name] = r
await self._save_repos()
return r
def get_repo(self, name: str) -> Repo:
"""
Returns a repo object with the given name.
:param name: Repo name
:return: Repo object or None
"""
return self._repos.get(name, None)
def get_all_repo_names(self) -> Tuple[str]:
"""
Returns a tuple of all repo names
:return:
"""
# noinspection PyTypeChecker
return tuple(self._repos.keys())
async def delete_repo(self, name: str):
"""
Deletes a repo and its' folders with the given name.
:param name:
:return:
"""
repo = self.get_repo(name)
if repo is None:
raise MissingGitRepo("There is no repo with the name {}".format(name))
shutil.rmtree(str(repo.folder_path))
repos = self.downloader_config.repos()
try:
del self._repos[name]
except KeyError:
pass
await self._save_repos()
async def update_all_repos(self) -> MutableMapping[Repo, Tuple[str, str]]:
"""
Calls repo.update() on all repos, returns a mapping of repos
that received new commits to a tuple containing old and
new commit hashes.
:return:
"""
ret = {}
for _, repo in self._repos.items():
old, new = await repo.update()
if old != new:
ret[repo] = (old, new)
await self._save_repos()
return ret
def _load_repos(self) -> MutableMapping[str, Repo]:
return {
name: Repo.from_json(data) for name, data in
self.downloader_config.repos().items()
}
async def _save_repos(self):
repo_json_info = {name: r.to_json() for name, r in self._repos.items()}
await self.downloader_config.set("repos", repo_json_info)

2
cogs/downloader/repos/.gitignore vendored Normal file
View File

@ -0,0 +1,2 @@
*
!.gitignore

0
tests/cogs/__init__.py Normal file
View File

0
tests/cogs/downloader/__init__.py Normal file
View File

136
tests/cogs/downloader/test_downloader.py Normal file
View File

@ -0,0 +1,136 @@
from collections import namedtuple
from raven.versioning import fetch_git_sha
import pytest
from cogs.downloader.repo_manager import RepoManager, Repo
from pathlib import Path
async def fake_run(*args, **kwargs):
fake_result_tuple = namedtuple("fake_result", "returncode result")
res = fake_result_tuple(0, (args, kwargs))
print(args[0])
return res
async def fake_run_noprint(*args, **kwargs):
fake_result_tuple = namedtuple("fake_result", "returncode result")
res = fake_result_tuple(0, (args, kwargs))
return res
@pytest.fixture(scope="module", autouse=True)
def patch_relative_to(monkeysession):
def fake_relative_to(self, some_path: Path):
return self
monkeysession.setattr("pathlib.Path.relative_to", fake_relative_to)
@pytest.fixture(scope="module")
def repo_manager(tmpdir_factory, config):
config.register_global(repos={})
rm = RepoManager(config)
rm.repos_folder = Path(str(tmpdir_factory.getbasetemp())) / 'repos'
return rm
@pytest.fixture
def repo(tmpdir):
from cogs.downloader.repo_manager import Repo
repo_folder = Path(str(tmpdir)) / 'repos' / 'squid'
repo_folder.mkdir(parents=True, exist_ok=True)
return Repo(
url="https://github.com/tekulvw/Squid-Plugins",
name="squid",
branch="rewrite_cogs",
folder_path=repo_folder
)
@pytest.fixture
def repo_norun(repo):
repo._run = fake_run
return repo
@pytest.fixture
def bot_repo(event_loop):
cwd = Path.cwd()
return Repo(
name="Red-DiscordBot",
branch="WRONG",
url="https://empty.com/something.git",
folder_path=cwd,
loop=event_loop
)
def test_existing_git_repo(tmpdir):
from cogs.downloader.repo_manager import Repo
repo_folder = Path(str(tmpdir)) / 'repos' / 'squid' / '.git'
repo_folder.mkdir(parents=True, exist_ok=True)
r = Repo(
url="https://github.com/tekulvw/Squid-Plugins",
name="squid",
branch="rewrite_cogs",
folder_path=repo_folder.parent
)
exists, _ = r._existing_git_repo()
assert exists is True
@pytest.mark.asyncio
async def test_clone_repo(repo_norun, capsys):
await repo_norun.clone()
clone_cmd, _ = capsys.readouterr()
clone_cmd = clone_cmd.strip('[\']').split('\', \'')
assert clone_cmd[0] == 'git'
assert clone_cmd[1] == 'clone'
assert clone_cmd[2] == '-b'
assert clone_cmd[3] == 'rewrite_cogs'
assert clone_cmd[4] == repo_norun.url
assert 'repos/squid' in clone_cmd[5]
@pytest.mark.asyncio
async def test_add_repo(monkeypatch, repo_manager):
monkeypatch.setattr("cogs.downloader.repo_manager.Repo._run",
fake_run_noprint)
squid = await repo_manager.add_repo(
url="https://github.com/tekulvw/Squid-Plugins",
name="squid",
branch="rewrite_cogs"
)
assert squid.available_modules == []
@pytest.mark.asyncio
async def test_current_branch(bot_repo):
branch = await bot_repo.current_branch()
# So this does work, just not sure how to fully automate the test
assert branch not in ("WRONG", "")
@pytest.mark.asyncio
async def test_current_hash(bot_repo):
branch = await bot_repo.current_branch()
bot_repo.branch = branch
commit = await bot_repo.current_commit()
sentry_sha = fetch_git_sha(str(bot_repo.folder_path))
assert sentry_sha == commit

70
tests/cogs/downloader/test_installable.py Normal file
View File

@ -0,0 +1,70 @@
import pytest
import json
from cogs.downloader.installable import Installable, InstallableType
from pathlib import Path
INFO_JSON = {
"author": (
"tekulvw",
),
"bot_version": (3, 0, 0),
"description": "A long description",
"hidden": False,
"install_msg": "A post-installation message",
"required_cogs": {},
"requirements": (
"tabulate"
),
"short": "A short description",
"tags": (
"tag1",
"tag2"
),
"type": "COG"
}
@pytest.fixture
def installable(tmpdir):
cog_path = tmpdir.mkdir("test_repo").mkdir("test_cog")
info_path = cog_path.join("info.json")
info_path.write_text(json.dumps(INFO_JSON), 'utf-8')
cog_info = Installable(Path(str(cog_path)))
return cog_info
def test_process_info_file(installable):
for k, v in INFO_JSON.items():
if k == "type":
assert installable.type == InstallableType.COG
else:
assert getattr(installable, k) == v
# noinspection PyProtectedMember
def test_location_is_dir(installable):
assert installable._location.exists()
assert installable._location.is_dir()
# noinspection PyProtectedMember
def test_info_file_is_file(installable):
assert installable._info_file.exists()
assert installable._info_file.is_file()
def test_name(installable):
assert installable.name == "test_cog"
def test_repo_name(installable):
assert installable.repo_name == "test_repo"
def test_serialization(installable):
data = installable.to_json()
location = data["location"]
assert location[-1] == "test_cog"

8
tests/conftest.py
View File

@ -5,10 +5,18 @@ import pytest
import random
from core.bot import Red
from _pytest.monkeypatch import MonkeyPatch
from core.drivers import red_json
from core import Config
@pytest.fixture(scope="session")
def monkeysession(request):
mpatch = MonkeyPatch()
yield mpatch
mpatch.undo()
@pytest.fixture(scope="module")
def json_driver(tmpdir_factory):
driver = red_json.JSON(