Commit e2b4a33b by Francisco Huertas

Add retry, log of result and force options in clone

Include test AT
Fix boot parameters
Enable password via parameter
parent 8d238405
...@@ -27,6 +27,6 @@ for PACKAGE in ${PACKAGES[@]}; do ...@@ -27,6 +27,6 @@ for PACKAGE in ${PACKAGES[@]}; do
${TEST_DIR}/bin/nosetests --verbosity=${VERBOSITY} --with-coverage --cover-package=${PACKAGE} --match="_it\b" ${TEST} ${TEST_DIR}/bin/nosetests --verbosity=${VERBOSITY} --with-coverage --cover-package=${PACKAGE} --match="_it\b" ${TEST}
done done
done done
$TEST_DIR/bin/coverage xml ${TEST_DIR}/bin/coverage xml
echo "Total report" echo "Total report"
$TEST_DIR/bin/coverage report ${TEST_DIR}/bin/coverage report
"""Naval Fate. """Naval Fate.
Usage: Usage:
git-cloner clon [(-u <user> | --user=<user>)] [-f] [(-o <out>| --out=<out>)] git-cloner clon [(-u <user> | --user=<user>) [(-p <pass>| --pass=<pass>)]] [(-f | --force)] [(-V | --verbose)] [(-o <out>| --out=<out>)]
(-c <org> | --org=<org>) (-l <limit> | --limit=<limit>) [(-r <file>| --retry=<file>)] ((-c <org> | --org=<org>) | (--U <user> | --target-user=<user>))
(-t <type> | --type=<type>)
[(-s <file> | --save=<file>)] [--ssh] [(-s <file> | --save=<file>)] [--ssh]
git-cloner make-repos [(-u <user> | --user=<user>)] [-f] [(-i <in>| --in=<in>)] [(-s <file> | --save=<file>)] git-cloner make-repos [(-u <user> | --user=<user>)] [-f] [(-i <in>| --in=<in>)] [(-s <file> | --save=<file>)]
(-t <type> | --type=<type>) --priv-token=<token> (-h | --host=<host>) [--ssh] (-t <type> | --type=<type>) --priv-token=<token> (-h | --host=<host>) [--ssh]
...@@ -12,35 +13,48 @@ Usage: ...@@ -12,35 +13,48 @@ Usage:
Options: Options:
-l --limit=<limit> Limit of repositories -l --limit=<limit> Limit of repositories
-u --user=<user> User to connect -u --user=<user> User to connect
-p --pass=<pass> Password to connect. The prompt ask if it is not provided
-v --version Show version. -v --version Show version.
-c --org=<org> Origin to read repositories -V --verbose Show version.
-c --org=<org> Origin (organization) to read repositories. It is incompatible with target-user
-h --help Show help -h --help Show help
-f Force remove directories -f --force Force remove directories
-o --out=<out> Output directory -o --out=<out> Output directory
-i --in=<in> Input directory -i --in=<in> Input directory
-s --save=<file> Save Output resutl into a file -s --save=<file> Save Output result into a file
-r --retry=<file> Retry the for the last execution, the input is te save file from the previous execution
--ssh Use ssh instead https --ssh Use ssh instead https
-t --type=<type> Indicate the type of repository gitlab is only supported jet -t --type=<type> Indicate the type of repository gitlab is only supported jet
-U --target-user=<user> Origin (user) to read repositories. It is incompatible with org
--priv-token=<token> Indicate the path to the private token where the token is --priv-token=<token> Indicate the path to the private token where the token is
-h --host=<host> Indicate the host. i.e git@gitlab.com:4123 -H --host=<host> Indicate the host. i.e git@gitlab.com:4123
""" """
from docopt import docopt from docopt import docopt
from git_cloner.github_wrapper import clones from git_cloner.operations.mass_cloner import mass_cloner
from git_cloner.github_wrapper import list_repos from git_cloner.utils.logger import get_logger_console
from git_cloner.utils.config import Config
from git_cloner.utils.utils import * from git_cloner.utils.utils import *
def main(): def main():
version = read('VERSION') arguments = docopt(__doc__, version='Git cloner {version}'.format(version='1.1.0-SNAPSHOT'))
arguments = docopt(__doc__, version='Git cloner {version}'.format(version=version))
run(arguments) run(arguments)
logger = get_logger_console(__name__)
def run(arguments): def run(arguments):
logger.debug(arguments)
Config.load_config(arguments) Config.load_config(arguments)
repos = list_repos(Config.get_config().org, Config.get_config().limit) if arguments['clon']:
clones(repos) mass_cloner()
# repos = list_repos(Config.get_config().org, Config.get_config().limit)
# clones(repos)
if __name__ == '__main__':
main()
...@@ -34,11 +34,6 @@ def list_repos_page(org, page): ...@@ -34,11 +34,6 @@ def list_repos_page(org, page):
print(len(filter_repos)) print(len(filter_repos))
return filter_repos return filter_repos
# return [{'name': 'dg-agent-commons', 'clone_url': 'https://github.com/Stratio/dg-agent-commons.git',
# 'ssh_url': 'git@github.com:Stratio/dg-agent-commons.git'},
# {'name': 'tiki-takka', 'clone_url': 'https://github.com/Stratio/tiki-takka.git',
# 'ssh_url': 'git@github.com:Stratio/tiki-takka.git'}]
def clones(repos): def clones(repos):
out_dir = '.' if Config.get_config().dir_out is None else Config.get_config().dir_out out_dir = '.' if Config.get_config().dir_out is None else Config.get_config().dir_out
......
from git_cloner.model.owner.owner import Owner from git_cloner.model.owner.owner import Owner
from git_cloner.model.repository import RawRepository from git_cloner.model.repository import RawRepository
from git_cloner.utils.config import Config
from git_cloner.utils import utils from git_cloner.utils import utils
from git_cloner.utils.config import Config
class GithubOwner(Owner): class GithubOwner(Owner):
......
...@@ -30,3 +30,6 @@ class GithubRemote(Remote): ...@@ -30,3 +30,6 @@ class GithubRemote(Remote):
def get_name(self): def get_name(self):
return self.name return self.name
def get_owner(self):
return self.owner
from uuid import uuid4 from uuid import uuid4
from git_cloner.model.remotes.remote import Remote from git_cloner.model.remotes.remote import Remote
...@@ -29,5 +30,3 @@ class GitlabRemote(Remote): ...@@ -29,5 +30,3 @@ class GitlabRemote(Remote):
'http': self.HTTP_URL_BASE.format(host=self.host, owner=self.owner, repo=self.repo_name), 'http': self.HTTP_URL_BASE.format(host=self.host, owner=self.owner, repo=self.repo_name),
'https': self.HTTPS_URL_BASE.format(host=self.host, owner=self.owner, repo=self.repo_name), 'https': self.HTTPS_URL_BASE.format(host=self.host, owner=self.owner, repo=self.repo_name),
}.get(self.type) }.get(self.type)
...@@ -7,6 +7,9 @@ class Remote(object): ...@@ -7,6 +7,9 @@ class Remote(object):
def get_name(self): def get_name(self):
raise NotImplementedError("Please Implement this method") raise NotImplementedError("Please Implement this method")
def get_owner(self):
raise NotImplementedError("Please Implement this method")
def build_remote_url(self): def build_remote_url(self):
raise NotImplementedError("Please Implement this method") raise NotImplementedError("Please Implement this method")
......
from pathlib import Path
import subprocess import subprocess
from pathlib import Path
import copy import copy
import os import os
import shutil
from git_cloner.model.remotes.github_remote import GithubRemote from git_cloner.model.remotes.github_remote import GithubRemote
from git_cloner.utils.config import Config from git_cloner.utils.logger import *
class Repository(object): class Repository(object):
...@@ -16,15 +16,27 @@ class Repository(object): ...@@ -16,15 +16,27 @@ class Repository(object):
GITLAB = "gitlab" GITLAB = "gitlab"
GITHUB = "github" GITHUB = "github"
OPERATION_CLONE = "clone"
@staticmethod @staticmethod
def clone(remote, root_path): def clone(remote, root_path):
logger = get_logger_console(Repository)
new_remote = copy.copy(remote) new_remote = copy.copy(remote)
new_remote.name = "origin" new_remote.name = "origin"
path = os.path.join(root_path, new_remote.repo_name) clone_path = os.path.join(root_path, new_remote.repo_name)
result = subprocess.run(new_remote.build_clone_cmd().format(path=path).split(" ")) if Config.get_config().force:
shutil.rmtree(clone_path, ignore_errors=True)
elif os.path.exists(clone_path):
logger.error("Error cloning: {}, path exists {}".format(remote.build_remote_url(), clone_path))
log_result(remote, Repository.OPERATION_CLONE, False)
return None
result = subprocess.run(new_remote.build_clone_cmd().format(path=clone_path).split(" "))
if result.returncode != 0: if result.returncode != 0:
raise RuntimeError("Error cloning: {}".format(remote.build_remote_url())) logger.error("Error cloning: {}".format(remote.build_remote_url()))
return Repository(path, [new_remote]) log_result(remote, Repository.OPERATION_CLONE, False)
return None
log_result(remote, Repository.OPERATION_CLONE, True)
return Repository(clone_path, [new_remote])
def __init__(self, path, remotes=None): def __init__(self, path, remotes=None):
if remotes is None: if remotes is None:
......
import json
from git_cloner.model.owner.owner import Owner from git_cloner.model.owner.owner import Owner
from git_cloner.model.repository import RawRepository
from git_cloner.utils.config import Config from git_cloner.utils.config import Config
def mass_cloner(): def mass_cloner():
owner = Owner.builder(Config.get_config().type)
out = Config.get_config().dir_out out = Config.get_config().dir_out
repos = owner.get_repos() repos = get_repos()
[repo.clone(out) for repo in repos] results = [repo.clone(out) for repo in repos]
return [result for result in results if result is not None]
def get_repos():
if Config.get_config().retry is None:
owner = Owner.builder(Config.get_config().type)
return owner.get_repos()
return log_retry_path(Config.get_config().retry)
def log_retry_path(path):
template_dic = """{{"name": "{name}","owner": {{"login": "{login}"}}}}"""
with open(path) as file:
lines = [line.strip() for line in file.readlines()]
filtered = [line for line in lines if line.endswith("False")]
format_str = [template_dic.format(
name=line.split(",")[0].split(":")[1].split("/")[1].strip(),
login=line.split(",")[0].split(":")[1].split("/")[0].strip())
for line in filtered]
return [RawRepository(json.loads(line)) for line in format_str]
...@@ -20,13 +20,16 @@ class Config(object): ...@@ -20,13 +20,16 @@ class Config(object):
_KEY_LIMIT = "--limit" _KEY_LIMIT = "--limit"
_KEY_DIR_IN = "--in" _KEY_DIR_IN = "--in"
_KEY_DIR_OUT = "--out" _KEY_DIR_OUT = "--out"
_KEY_FORCE = "-f" _KEY_FORCE = "--force"
_KEY_SAVE = "--save" _KEY_SAVE = "--save"
_KEY_SSH = "--ssh" _KEY_SSH = "--ssh"
_KEY_VERBOSE = "--verbose"
_KEY_RETRY = "--retry"
instance = None instance = None
def __init__(self, user, password, org, limit, dir_in, dir_out, def __init__(self, user, password, org, limit, dir_in, dir_out,
force, save, ssh, repo_type, token, host, target_user): force, save, ssh, repo_type, token, host, target_user, verbose, retry):
self.retry = retry
self.target_user = target_user self.target_user = target_user
self.user = user self.user = user
self.password = password self.password = password
...@@ -40,6 +43,7 @@ class Config(object): ...@@ -40,6 +43,7 @@ class Config(object):
self.type = repo_type self.type = repo_type
self.token = token self.token = token
self.host = host self.host = host
self.verbose = verbose
@classmethod @classmethod
def load_config(cls, arguments): def load_config(cls, arguments):
...@@ -54,15 +58,17 @@ class Config(object): ...@@ -54,15 +58,17 @@ class Config(object):
repo_type = arguments.pop(cls._KEY_TYPE, None) repo_type = arguments.pop(cls._KEY_TYPE, None)
host = arguments.pop(cls._KEY_HOST, None) host = arguments.pop(cls._KEY_HOST, None)
path = arguments.pop(cls._KEY_TOKEN, None) path = arguments.pop(cls._KEY_TOKEN, None)
verbose = arguments.pop(cls._KEY_VERBOSE, False)
token = read(path) if path is not None else None token = read(path) if path is not None else None
target_user = arguments.pop(cls._KEY_TARGET_USER, None) target_user = arguments.pop(cls._KEY_TARGET_USER, None)
retry = arguments.pop(cls._KEY_RETRY, None)
if user is not None and password is None: if user is not None and password is None:
password = getpass.getpass() password = getpass.getpass()
cls.instance = Config(user, password, org, 100, dir_in, dir_out, cls.instance = Config(user, password, org, 100, dir_in, dir_out,
force, save, ssh, repo_type, token, host, target_user) force, save, ssh, repo_type, token, host, target_user, verbose, retry)
return cls.instance return cls.instance
@classmethod @classmethod
def get_config(cls): def get_config(cls):
return cls.instance return cls.instance if cls.instance is not None else cls.load_config({})
import logging
from git_cloner.utils.config import Config
def get_logger_console(origin_module,
str_format='%(asctime)s %(name)-12s %(levelname)-8s %(message)s',
date_format='%m-%d %H:%M'):
level = logging.DEBUG if Config.get_config().verbose else logging.INFO
logging.basicConfig(
level=level,
format=str_format,
datefmt=date_format)
logger_name = origin_module if isinstance(origin_module, str) else origin_module.__module__
return logging.getLogger(logger_name)
def log_result(remote, operation, result):
template = "\nRepo: {}/{}, Operation: {}, Result: {}"
if Config.get_config().save is not None:
with open(Config.get_config().save, "a") as my_file:
my_file.write(template.format(remote.owner, remote.repo_name, operation, result))
...@@ -16,6 +16,6 @@ def make_request(url, headers=None, auth_if_possible=True): ...@@ -16,6 +16,6 @@ def make_request(url, headers=None, auth_if_possible=True):
auth = (config.user, config.password) if config.user and auth_if_possible else None auth = (config.user, config.password) if config.user and auth_if_possible else None
result = requests.get(url, auth=auth, headers=headers) result = requests.get(url, auth=auth, headers=headers)
if result.status_code != 200: if result.status_code != 200:
raise RuntimeError("Incorrect request {}\n{}".format(url,str(result.content))) raise RuntimeError("Incorrect request {}\n{}".format(url, str(result.content)))
return result.json() return result.json()
...@@ -2,7 +2,7 @@ import os ...@@ -2,7 +2,7 @@ import os
from setuptools import setup from setuptools import setup
def read(fname): def setup_read(fname):
return open(os.path.join(os.path.dirname(__file__), fname)).read().strip() return open(os.path.join(os.path.dirname(__file__), fname)).read().strip()
...@@ -10,17 +10,17 @@ required = [ ...@@ -10,17 +10,17 @@ required = [
'docopt==0.6.2', 'docopt==0.6.2',
'requests==2.18.4' 'requests==2.18.4'
] ]
os.path.join(os.getcwd(),'VERSION')
setup(name='git_cloner', setup(name='git_cloner',
version=read('VERSION'), version=setup_read('VERSION'),
author="Francisco Huertas", author="Francisco Huertas",
author_email="francisco@fhuertas.com", author_email="francisco@fhuertas.com",
license="Apache2", license="Apache2",
packages=["git_cloner"], packages=["git_cloner"],
description="Python tool for clon repositories", description="Python tool for clon repositories",
long_description=read('README.md'), long_description=setup_read('README.md'),
url='https://github.com/fhuertas/python_base', url='https://github.com/fhuertas/python_base',
download_url="https://github.com/fhuertas/python_base/tarball/{}".format(read('VERSION')), download_url="https://github.com/fhuertas/python_base/tarball/{}".format(setup_read('VERSION')),
install_requires=required, install_requires=required,
classifiers=[ classifiers=[
"Development Status :: 5 - Production/Stable", "Development Status :: 5 - Production/Stable",
...@@ -31,6 +31,6 @@ setup(name='git_cloner', ...@@ -31,6 +31,6 @@ setup(name='git_cloner',
], ],
entry_points={ entry_points={
'console_scripts': [ 'console_scripts': [
'git-cloner=git_cloner.boot:main', 'git-cloner=git_cloner.boot.boot:main',
], ],
},) },)
"""{"--force": False,
"--help": False,
"--host": None,
"--in": None,
"--org": "stratio",
"--out": None,
"--priv-token": None,
"--retry": None,
"--save": None,
"--ssh": True,
"--type": "github",
"--user": None,
"--verbose": False,
"--version": False,
"clon": True,
"make-repos": False}"""
import unittest
import runpy
import sys
from git_cloner.boot import boot
from git_cloner.utils.utils import read
from tests.git_cloner.utils.utils import *
from pkg_resources import resource_filename
test_dir = "/tmp/test_tmp"
secrets_path = os.path.join(resource_filename('tests', 'resources'), 'secrets')
class BootAT(unittest.TestCase):
def setUp(self):
self.old_dir = set_test_dir(test_dir)
self.args = sys.argv
def tearDown(self):
sys.argv = self.args
restore_path(self.old_dir)
# Command: git-cloner clon -u user -p pass -U yunxao --ssh -t github -o /tmp/test_tmp/fhuertas -s /tmp/test_tmp/log.log
def test_run_clone_at(self):
print("Command")
print("git-cloner clon -u user -p pass -U yunxao --ssh -t github -o /tmp/test_tmp/fhuertas "
"-s /tmp/test_tmp/log.log")
user = read(os.path.join(secrets_path, 'user'))
password = read(os.path.join(secrets_path, 'password'))
cmd = "git-cloner clon -u {user} -p {password} -U yunxao --ssh -t github -o {out_dir}/fhuertas " \
"-s {out_dir}/log.log".format(out_dir=test_dir, user=user, password=password)
sys.argv = cmd.split(" ")
runpy.run_module(boot.__name__, run_name="__main__", alter_sys=False)
with open(os.path.join(test_dir, "log.log")) as file:
lines = len([line for line in file.readlines() if line.strip() != ""])
self.assertEqual(lines, 10)
import json import time
import unittest import unittest
import os import pkg_resources
import shutil from mock_decorators.function_mock import FunctionMock, FunctionMockResult
import time
from mock_decorators.function_mock import FunctionMock
from git_cloner.model.owner.github_owner import GithubOwner
from git_cloner.model.repository import Repository from git_cloner.model.repository import Repository
from git_cloner.operations import mass_cloner from git_cloner.operations.mass_cloner import *
from git_cloner.utils import utils from git_cloner.utils import utils
from git_cloner.utils.config import Config from git_cloner.utils.config import Config
import pkg_resources from tests.git_cloner.utils.gen_utils import generate_repo_github
from tests.git_cloner.utils.utils import *
test_dir = "/tmp/test_tmp" test_dir = "/tmp/test_tmp"
response_filepath = pkg_resources.resource_filename("tests", "resources/request_fhuertas_org.json") response_filepath = pkg_resources.resource_filename("tests", "resources/request_fhuertas_org.json")
log_retry_path = pkg_resources.resource_filename("tests", "resources/retry.log")
class MassClonerTest(unittest.TestCase): class MassClonerTest(unittest.TestCase):
def setUp(self): def setUp(self):
shutil.rmtree(test_dir, ignore_errors=True) self.dir = set_test_dir(test_dir)
os.makedirs(test_dir)
time.sleep(1) time.sleep(1)
def tearDown(self): def tearDown(self):
shutil.rmtree(test_dir, True) restore_path(self.dir)
time.sleep(1) time.sleep(1)
def serial_it(self): def serial_it(self):
...@@ -32,7 +32,7 @@ class MassClonerTest(unittest.TestCase): ...@@ -32,7 +32,7 @@ class MassClonerTest(unittest.TestCase):
config.type = "github" config.type = "github"
config.target_user = "yunxao" config.target_user = "yunxao"
config.dir_out = test_dir config.dir_out = test_dir
mass_cloner.mass_cloner() mass_cloner()
repos = [name for name in os.listdir(test_dir) if Repository(os.path.join(test_dir, name)).check_git()] repos = [name for name in os.listdir(test_dir) if Repository(os.path.join(test_dir, name)).check_git()]
self.assertEqual(len(repos), 10) self.assertEqual(len(repos), 10)
...@@ -42,7 +42,7 @@ class MassClonerTest(unittest.TestCase): ...@@ -42,7 +42,7 @@ class MassClonerTest(unittest.TestCase):
config.org = "fhuertasorgfortest" config.org = "fhuertasorgfortest"
config.ssh = True config.ssh = True
config.dir_out = test_dir config.dir_out = test_dir
mass_cloner.mass_cloner() mass_cloner()
repos = [name for name in os.listdir(test_dir) if Repository(os.path.join(test_dir, name)).check_git()] repos = [name for name in os.listdir(test_dir) if Repository(os.path.join(test_dir, name)).check_git()]
self.assertEqual(len(repos), 1) self.assertEqual(len(repos), 1)
...@@ -59,15 +59,82 @@ class MassClonerTest(unittest.TestCase): ...@@ -59,15 +59,82 @@ class MassClonerTest(unittest.TestCase):
@FunctionMock(utils, 'make_request', mock_request, False) @FunctionMock(utils, 'make_request', mock_request, False)
def inner_test(): def inner_test():
mass_cloner.mass_cloner() mass_cloner()
inner_test() inner_test()
repos = [name for name in os.listdir(test_dir) if Repository(os.path.join(test_dir, name)).check_git()] repos = [name for name in os.listdir(test_dir) if Repository(os.path.join(test_dir, name)).check_git()]
self.assertEqual(len(repos), 1) self.assertEqual(len(repos), 1)
def test_logger():
config = Config.load_config({})
config.type = "github"
config.org = "fhuertasorgfortest"
config.ssh = True
config.save = os.path.join(test_dir, "log.log")
config.dir_out = test_dir
repos_raw = [
generate_repo_github(name="monkey", user="fhuertasorgfortest"),
generate_repo_github()]
repos = [RawRepository(repo) for repo in repos_raw]
@FunctionMockResult(entity=GithubOwner, function_name="get_repos", result=repos)
def inner_test():
mass_cloner()
inner_test()
with open(config.save) as file:
result = "".join(file.readlines()).strip()
self.assertEqual(result,
"Repo: fhuertasorgfortest/monkey, Operation: clone, Result: True\n"
"Repo: {}, Operation: clone, Result: False".format(repos_raw[1].get('full_name')))
def test_get_repos_normal():
config = Config.load_config({})
config.type = "github"
config.org = "fhuertasorgfortest"
config.ssh = True
result = get_repos()
self.assertEqual(len(result), 1)
def test_get_repos_retry():
config = Config.load_config({})
config.type = "github"
config.org = "fhuertasorgfortest"
config.ssh = True
config.retry = log_retry_path
result = get_repos()
self.assertEqual(len(result), 2)
self.assertEqual(result[0].raw_data['name'], 'monkey')
self.assertEqual(result[0].raw_data['owner']['login'], 'fhuertasorgfortest')
self.assertEqual(result[1].raw_data['name'], 'mix')
self.assertEqual(result[1].raw_data['owner']['login'], 'mix')
def test_retry_mass_clone():
config = Config.load_config({})
config.type = "github"
config.org = "fhuertasorgfortest"
config.ssh = True
config.retry = log_retry_path
result = mass_cloner()
self.assertEqual(len(result), 1)
self.setUp() self.setUp()
test_mass_cloner_user() test_mass_cloner_user()
self.tearDown()
self.setUp() self.setUp()
test_mass_cloner_org() test_mass_cloner_org()
self.tearDown()
self.setUp() self.setUp()
test_mass_cloner_fail() test_mass_cloner_fail()
self.setUp()
self.tearDown()
test_logger()
test_get_repos_normal()
test_get_repos_retry()
self.tearDown()
self.setUp()
test_retry_mass_clone()
monkey @ 7b68cc95
Subproject commit 7b68cc95ef2a31611b76b6146cdee7bb7413f013
import subprocess
import unittest import unittest
import os from mock_decorators.function_mock import FunctionMock
import shutil from mock_decorators.object_mock import ObjectMock
from git_cloner.model.remotes.github_remote import GithubRemote from git_cloner.model.remotes.github_remote import GithubRemote
from git_cloner.model.repository import Repository, RawRepository from git_cloner.model.repository import Repository, RawRepository
from git_cloner.utils.config import Config from git_cloner.utils.config import Config
from tests.git_cloner.utils.gen_utils import * from tests.git_cloner.utils.gen_utils import *
from tests.git_cloner.utils.utils import *
test_dir = "/tmp/test_tmp" test_dir = "/tmp/test_tmp"
class RepositoryIT(unittest.TestCase): class RepositoryIT(unittest.TestCase):
def setUp(self): def setUp(self):
shutil.rmtree(test_dir, ignore_errors=True) self.old_dir = set_test_dir(test_dir)
os.makedirs(test_dir) os.makedirs(test_dir)
def tearDown(self): def tearDown(self):
shutil.rmtree(test_dir, True) restore_path(self.old_dir) # , test_dir)
# This tests have problems if are executed in parallel # This tests have problems if are executed in parallel
def serials_it(self): def serials_it(self):
def creating_with_cloning(): def creating_with_cloning():
remote = GithubRemote.build_from_url("https://github.com/fhuertas/scala_base.git") remote = GithubRemote.build_from_url("https://github.com/fhuertas/scala_base.git")
repo = Repository.clone(remote, test_dir) repo = Repository.clone(remote, test_dir)
...@@ -52,3 +53,38 @@ class RepositoryIT(unittest.TestCase): ...@@ -52,3 +53,38 @@ class RepositoryIT(unittest.TestCase):
creating_with_cloning() creating_with_cloning()
creating_with_cloning_ssh() creating_with_cloning_ssh()
clone_repo_github() clone_repo_github()
def check_force_remove_dir_it(self):
remote = GithubRemote.build_from_url("https://github.com/fhuertas/scala_base.git")
target_path = os.path.join(test_dir, remote.repo_name)
os.mkdir(target_path)
Config.load_config({'--force': True})
def clone_subprocess_function(*args, **kwargs):
os.makedirs(target_path)
return_process = ObjectMock
return_process.returncode = 0
return return_process
@FunctionMock(entity=subprocess, function_name="run", mocked_function=clone_subprocess_function,
check_signature=False)
def inner_test():
return Repository.clone(remote, test_dir)
result = inner_test()
self.assertTrue(os.path.exists(target_path))
self.assertIsNotNone(result)
def check_no_force_remove_dir_it(self):
remote = GithubRemote.build_from_url("https://github.com/fhuertas/scala_base.git")
target_path = os.path.join(test_dir, remote.repo_name)
os.mkdir(target_path)
Config.load_config({'--force': False})
# Config.load_config({'--force': False, '--save': target_path_log})
result = Repository.clone(remote, test_dir)
self.assertIsNone(result)
self.assertTrue(os.path.exists(target_path))
import subprocess
import unittest
import os
import shutil
from mock_decorators.function_mock import FunctionMockCheckCall
from git_cloner.model.repository import Repository
test_dir = "/tmp/test_tmp"
class TestRepository(unittest.TestCase):
def test_add_remote(self):
pass
def setUp(self):
shutil.rmtree(test_dir, True)
os.makedirs(test_dir, exist_ok=True)
def tearDown(self):
shutil.rmtree(test_dir, True)
@FunctionMockCheckCall(subprocess, "run", expected_times=2)
def test_init_git_and_check(self):
rep_dir = "/tmp/test_tmp/git"
os.makedirs(rep_dir, exist_ok=True)
rep = Repository("/tmp/test_tmp/git")
rep.init_git()
self.assertTrue(rep.check_git())
shutil.rmtree(rep_dir, True)
def test_git_check_normal_dir(self):
rep = Repository("/tmp/test_tmp")
self.assertFalse(rep.check_git())
pass
...@@ -3,9 +3,8 @@ import unittest ...@@ -3,9 +3,8 @@ import unittest
from mock_decorators.function_mock import FunctionMockResult from mock_decorators.function_mock import FunctionMockResult
from git_cloner.utils import utils
from git_cloner.utils.config import Config
from git_cloner.utils import config from git_cloner.utils import config
from git_cloner.utils.config import Config
class TestConfig(unittest.TestCase): class TestConfig(unittest.TestCase):
...@@ -24,7 +23,9 @@ class TestConfig(unittest.TestCase): ...@@ -24,7 +23,9 @@ class TestConfig(unittest.TestCase):
Config._KEY_TOKEN: "token_path", Config._KEY_TOKEN: "token_path",
Config._KEY_TYPE: "type", Config._KEY_TYPE: "type",
Config._KEY_HOST: "git.host.com", Config._KEY_HOST: "git.host.com",
Config._KEY_TARGET_USER: "pepe" Config._KEY_TARGET_USER: "pepe",
Config._KEY_VERBOSE: True,
Config._KEY_RETRY: "/tmp/log"
} }
my_config = Config.load_config(arguments) my_config = Config.load_config(arguments)
...@@ -41,6 +42,8 @@ class TestConfig(unittest.TestCase): ...@@ -41,6 +42,8 @@ class TestConfig(unittest.TestCase):
self.assertEqual(my_config.token, "THIS_IS_A_TOKEN") self.assertEqual(my_config.token, "THIS_IS_A_TOKEN")
self.assertEqual(my_config.host, "git.host.com") self.assertEqual(my_config.host, "git.host.com")
self.assertEqual(my_config.target_user, "pepe") self.assertEqual(my_config.target_user, "pepe")
self.assertEqual(my_config.verbose, True)
self.assertEqual(my_config.retry, "/tmp/log")
def test_no_arguments(self): def test_no_arguments(self):
arguments = {} arguments = {}
...@@ -58,3 +61,5 @@ class TestConfig(unittest.TestCase): ...@@ -58,3 +61,5 @@ class TestConfig(unittest.TestCase):
self.assertEqual(my_config.token, None) self.assertEqual(my_config.token, None)
self.assertEqual(my_config.host, None) self.assertEqual(my_config.host, None)
self.assertEqual(my_config.target_user, None) self.assertEqual(my_config.target_user, None)
self.assertEqual(my_config.verbose, False)
self.assertEqual(my_config.retry, None)
...@@ -25,9 +25,7 @@ def gen_github_ssh_url(): ...@@ -25,9 +25,7 @@ def gen_github_ssh_url():
pass pass
def generate_repo_github(): def generate_repo_github(name=gen_name(), user=gen_name()):
name = gen_name()
user = gen_name()
full_path = join_path(user, name) full_path = join_path(user, name)
description = [gen_name() for _ in range(gen_int(limit=100))] description = [gen_name() for _ in range(gen_int(limit=100))]
api_url_base_repos = join_path("https://api.github.com/repos", full_path) api_url_base_repos = join_path("https://api.github.com/repos", full_path)
......
import os
import shutil
def set_test_dir(dir_path):
old_path = os.getcwd()
shutil.rmtree(dir_path, True)
os.makedirs(dir_path)
os.chdir(dir_path)
return old_path
def restore_path(old_path, dir_path=None):
if dir_path is not None:
shutil.rmtree(dir_path, True)
os.chdir(old_path)
Repo: fhuertasorgfortest/monkey, Operation: clone, Result: False
Repo: lower/mix, Operation: clone, Result: True
Repo: mix/mix, Operation: clone, Result: False
\ No newline at end of file
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment