Skip to content

P
python-cloner
Commit 8d238405 by Francisco Huertas
Options

Refactor structure

parent 127026c9
Showing with 1099 additions and 89 deletions
.gitignore
# IntelliJ files
.idea
python-cloner.iml
env
env-*
# Python temporal files
......@@ -12,3 +13,6 @@ __pycache__
# package files
dist
build
results
tests/resources/secrets
tests/resources/tmp
\ No newline at end of file
Makefile
PYTHON_VERSION=
PYTHON_ENV_TEST=env-test$(PYTHON_VERSION)
all: clean env test-versions package
all: clean env test package
env: env/bin/activate
env/bin/activate:
bin/env.sh
test:
test: test-ut test-it
test-ut:
bin/tests.sh $(PYTHON_VERSION)
test-clean:
rm -Rf env-*
test-it:
bin/tests_it.sh $(PYTHON_VERSION)
clean-test:
rm -Rf env-* .coverage coverage.xml
clean:
bin/clean.sh
......
VERSION
1.0.0
1.1.0-SNAPSHOT
bin/continuous-test.sh
......@@ -40,8 +40,8 @@ SOURCE_DIR='.'
# ficheros de test para buscar el test
TEST_DIR='.'
# prefijo y postfijo del test
PREFIX="test_"
POSTFIX=""
PREFIX=""
POSTFIX="_test"
# comando del test %s irá el nombre del test
TEST_COMMAND="python -m unittest %s"
# Filtro posix para ignorar ficheros. Este esta puesto para ignorar los ficheros temporales de intellij
......
bin/tests.sh
......@@ -5,14 +5,14 @@ source "${BASEDIR}/bin/var-env.sh"
PYTHON_VERSION=${PYTHON_VERSION:=${DEVELOP_PYTHON_VERSION}}
VERBOSITY=${VERBOSITY:=$TEST_VERBOSITY}
TEST_DIR=${BASEDIR}/env-test${PYTHON_VERSION}
TEST_DIR=${BASEDIR}/env-test-${PYTHON_VERSION}
echo "Testing on python version: ${PYTHON_VERSION}"
echo "Test environment: $TEST_DIR"
if [ ! -d "$TEST_DIR" ]; then
virtualenv -p python$1 -q $TEST_DIR
virtualenv -p python${PYTHON_VERSION} -q $TEST_DIR
echo "New virtualenv for UT created."
source $TEST_DIR/bin/activate
......
bin/tests_it.sh 0 → 100755
#!/bin/bash -e
BASEDIR=`dirname $0`/..
source "${BASEDIR}/bin/var-env.sh"
PYTHON_VERSION=${PYTHON_VERSION:=${DEVELOP_PYTHON_VERSION}}
VERBOSITY=${VERBOSITY:=$TEST_VERBOSITY}
TEST_DIR=${BASEDIR}/env-test-${PYTHON_VERSION}
echo "Testing on python version: ${PYTHON_VERSION}"
echo "Test environment: $TEST_DIR"
if [ ! -d "$TEST_DIR" ]; then
virtualenv -p python${PYTHON_VERSION} -q $TEST_DIR
echo "New virtualenv for IT created."
source $TEST_DIR/bin/activate
echo "New virtualenv for IT activated."
pip install -r $BASEDIR/requirements.txt
pip install -e $BASEDIR
fi
export COVERAGE_FILE=.coverage
for PACKAGE in ${PACKAGES[@]}; do
for TEST in ${TEST_FOLDERS[@]}; do
${TEST_DIR}/bin/nosetests --verbosity=${VERBOSITY} --with-coverage --cover-package=${PACKAGE} --match="_it\b" ${TEST}
done
done
$TEST_DIR/bin/coverage xml
echo "Total report"
$TEST_DIR/bin/coverage report
bin/var-env.sh
#!/usr/bin/env bash
export DEVELOP_PYTHON_VERSION=3
export TEST_PYTHON_VERSION=(2.7 3 3.4 3.5)
export TEST_PYTHON_VERSION=(3 3.4 3.5 3.6)
export TEST_VERBOSITY=3
export PACKAGES=("github_cloner")
export PACKAGES=("git_cloner")
export TEST_FOLDERS=("tests")
\ No newline at end of file
github_cloner/boot.py → git_cloner/boot/boot.py
......@@ -4,7 +4,8 @@ Usage:
git-cloner clon [(-u <user> | --user=<user>)] [-f] [(-o <out>| --out=<out>)]
(-c <org> | --org=<org>) (-l <limit> | --limit=<limit>)
[(-s <file> | --save=<file>)] [--ssh]
git-cloner make-repos [(-u <user> | --user=<user>)] [-f] [(-i <in>| --in=<in>)] [(-s <file> | --save=<file>)] (-t <type> | --type=<type>) --priv-token=<token>
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]
git-cloner (-h | --help)
git-cloner --version
......@@ -21,22 +22,25 @@ Options:
--ssh Use ssh instead https
-t --type=<type> Indicate the type of repository gitlab is only supported jet
--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
"""
from docopt import docopt
from github_cloner.config import Config
from github_cloner.github_wrapper import clones
from github_cloner.github_wrapper import list_repos
from git_cloner.github_wrapper import clones
from git_cloner.github_wrapper import list_repos
from git_cloner.utils.config import Config
from git_cloner.utils.utils import *
def main():
arguments = docopt(__doc__, version='Git cloner 1.0')
version = read('VERSION')
arguments = docopt(__doc__, version='Git cloner {version}'.format(version=version))
run(arguments)
def run(arguments):
Config.load_config(arguments)
repos = list_repos(Config.org, Config.limit)
repos = list_repos(Config.get_config().org, Config.get_config().limit)
clones(repos)
github_cloner/github_wrapper.py → git_cloner/github_wrapper.py
import pathlib
import shutil
import subprocess
import requests
import shutil
from github_cloner.config import Config
from git_cloner.utils.config import Config
def list_repos(org, limit):
print(limit)
limit = int(limit)
total_pages = int(limit / int(Config.limit_per_page)) + 1
if limit % Config.limit_per_page:
total_pages = int(limit / int(Config.get_config().limit_per_page)) + 1
if limit % Config.get_config().limit_per_page:
total_pages += 1
print(total_pages)
return [repo for page in range(1, total_pages) for repo in list_repos_page(org, page)]
......@@ -19,9 +19,9 @@ def list_repos(org, limit):
def list_repos_page(org, page):
KEYS = ['name', 'clone_url', 'ssh_url']
auth = (Config.user, Config.password) if Config.user else None
auth = (Config.get_config().user, Config.get_config().password) if Config.get_config().user else None
url = 'https://api.github.com/orgs/{org}/repos?page={page}&per_page={limit}'.format(
org=org, limit=Config.limit_per_page, page=page)
org=org, limit=Config.get_config().limit_per_page, page=page)
r = requests.get(url, auth=auth)
if r.status_code != 200:
print("Error {} in the request".format(r.status_code))
......@@ -41,22 +41,19 @@ def list_repos_page(org, page):
def clones(repos):
out_dir = '.' if Config.dir is None else Config.dir
force = Config.force
out_dir = '.' if Config.get_config().dir_out is None else Config.get_config().dir_out
force = Config.get_config().force
size = len(repos)
repo_result = []
if out_dir is not None:
print(Config.dir)
print(Config.dir)
print(Config.dir)
pathlib.Path(out_dir).mkdir(parents=True, exist_ok=True)
print("Cloning repositories {}".format(len(repos)))
n = 1
for repo in repos:
repo_dir = "{}/{}".format(out_dir, repo.get("name"))
repo_name = repo.get("name")
clone_url = repo.get('ssh_url') if Config.ssh else repo.get('clone_url')
clone_url = repo.get('ssh_url') if Config.get_config().ssh else repo.get('clone_url')
print("Clone repository {name} ({n}/{total})".format(n=n, total=size, name=repo_name))
cmd_str = 'git clone {clone_url} {dir}'.format(clone_url=clone_url, dir=repo_dir)
if force:
......@@ -72,7 +69,7 @@ def clones(repos):
print(result_str)
repo_result.append(result_str)
n += 1
if Config.save:
with open(Config.save, "w") as text_file:
if Config.get_config().save:
with open(Config.get_config().save, "w") as text_file:
text_file.write("\n".join(repo_result))
text_file.write("\n")
git_cloner/gitlab_wrapper.py 0 → 100644
import os
from pathlib import Path
from uuid import uuid4
from git_cloner import structure_repo
from git_cloner.utils.config import Config
def add_remote():
current_dir = Path.cwd()
remote = uuid4()
cmd_remote_add = "git remote add {remote} git://git@{host}/{org}/{name}".format(
name=current_dir.name,
remote=remote,
host=Config.get_config().host,
org=Config.get_config().org
)
print(cmd_remote_add)
return remote
def delete_remote(remote):
cmd_remote_remove = "git remote remove {remote}".format(remote=remote)
print(cmd_remote_remove)
def push_all(remote):
cmd_push = "git push --all {remote}".format(remote=remote)
print(cmd_push)
def get_namespace():
org = Config.get_config().org
url = "http://git.fhuertas.com:55080/api/v3/namespaces"
headers = {'PRIVATE-TOKEN': Config.get_config().token}
r = [{'id': 4, 'name': 'stratio', 'path': 'stratio', 'kind': 'group', 'full_path': 'stratio', 'parent_id': None,
'members_count_with_descendants': 1},
{'id': 3, 'name': 'personal', 'path': 'personal', 'kind': 'group', 'full_path': 'personal', 'parent_id': None,
'members_count_with_descendants': 1},
{'id': 2, 'name': 'fhuertas', 'path': 'fhuertas', 'kind': 'user', 'full_path': 'fhuertas', 'parent_id': None},
{'id': 1, 'name': 'root', 'path': 'root', 'kind': 'user', 'full_path': 'root', 'parent_id': None}]
# r = requests.get(url, headers=headers)
result = [e for e in r if e.get('name') == Config.get_config().org]
if len(result) > 0:
return result[0].get('id')
return -1
def create_repo(ns):
current_dir = Path.cwd()
cmd = 'curl --header "PRIVATE-TOKEN: {token}" -X POST "http://git.fhuertas.com:55080/api/v3/projects?name={' \
'name}&namespace_id={ns}"'.format(ns=ns, token=Config.get_config().token, name=current_dir.name)
print(cmd)
def push_repository(path, ns):
print("XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXxx")
current_dir = Path.cwd()
os.chdir(path)
print(Path.cwd())
remote = add_remote()
push_all(remote)
create_repo(ns)
delete_remote(remote)
# print(ns)
os.chdir(str(current_dir))
pass
def make_all():
path = Config.get_config().dir_in
ns = get_namespace()
print("NAMESPACE {}".format(ns))
dirs = structure_repo.list_repos(path)
[push_repository(path, ns) for path in dirs]
class GitlabWrapper(object):
pass
git_cloner/model/owner/github_owner.py 0 → 100644
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 import utils
class GithubOwner(Owner):
@staticmethod
def build_from_config():
conf = Config.get_config()
owner, is_user = (conf.target_user, True) if conf.target_user else (conf.org, False)
return GithubOwner(owner, is_user)
REPOS_FROM_USER_TEMPLATE = "{base}/repos?page={{p}}&per_page=100"
BASE_API_URL_USER = "https://api.github.com/users/{u}"
BASE_API_URL_ORG = "https://api.github.com/orgs/{o}"
def __init__(self, owner, is_user=True):
self.is_user = is_user
self.owner = owner
def api_base_url(self):
if self.is_user:
return self.BASE_API_URL_USER.format(u=self.owner)
return self.BASE_API_URL_ORG.format(o=self.owner)
def get_repos_url(self):
return self.REPOS_FROM_USER_TEMPLATE.format(base=self.api_base_url())
def get_repos(self):
return [RawRepository(repo) for repo in self.get_repos_rec(0, self.get_repos_url())]
# Is not the best form but i'm, trying tailrec
@staticmethod
def get_repos_rec(p, url, acc=list()):
limit = Config.get_config().limit
size = len(acc)
if p > 0 and size != limit * p:
return acc
else:
return GithubOwner.get_repos_rec(
p + 1, url,
acc + utils.make_request(url.format(p=p), auth_if_possible=True))
git_cloner/model/owner/owner.py 0 → 100644
class Owner(object):
def get_repos(self):
raise NotImplementedError("Please Implement this method")
@staticmethod
def build_from_config():
raise NotImplementedError("Please Implement this method")
@staticmethod
def builder(repo_type):
from git_cloner.model.owner.github_owner import GithubOwner
from git_cloner.model.repository import Repository
if repo_type == Repository.GITHUB: return GithubOwner.build_from_config()
raise RuntimeError("Invalid repository type")
git_cloner/model/remotes/github_remote.py 0 → 100644
from uuid import uuid4
from git_cloner.model.remotes.remote import Remote
class GithubRemote(Remote):
SSH_TEMPLATE = "git@github.com:{owner}/{repo_name}.git"
HTTPS_TEMPLATE = "https://github.com/{owner}/{repo_name}.git"
@staticmethod
def build_from_url(url):
name = url.split("/")[-1].replace(".git", "")
ssh = url.startswith("git")
if ssh:
owner = url.split(":")[-1].split("/")[0]
else:
owner = url.split("/")[-2]
return GithubRemote(owner, name, ssh)
def __init__(self, owner, repo_name, is_ssh, name=uuid4()):
self.ssh = is_ssh
self.name = name
self.repo_name = repo_name
self.owner = owner
def build_remote_url(self):
if self.ssh:
return self.SSH_TEMPLATE.format(owner=self.owner, repo_name=self.repo_name)
return self.HTTPS_TEMPLATE.format(owner=self.owner, repo_name=self.repo_name)
def get_name(self):
return self.name
git_cloner/model/remotes/gitlab_remote.py 0 → 100644
from uuid import uuid4
from git_cloner.model.remotes.remote import Remote
class GitlabRemote(Remote):
SSH_URL_BASE = "ssh://git@{host}/{owner}/{repo}"
HTTP_URL_BASE = "http://{host}/{owner}/{repo}"
HTTPS_URL_BASE = "https://{host}/{owner}/{repo}"
def __init__(self, host, owner, repo_name, remote_type, name=uuid4()):
self.name = name
self.host = host
self.owner = owner
self.repo_name = repo_name
self.type = remote_type
def get_name(self):
return self.name
def get_host(self):
return self.host
def get_owner(self):
return self.owner
def build_remote_url(self):
return {
'ssh': self.SSH_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),
}.get(self.type)
git_cloner/model/remotes/remote.py 0 → 100644
class Remote(object):
REMOTE_ADD_TEMPLATE = "git remote add {uuid} {url}"
REMOTE_DELETE_TEMPLATE = "git remote remove {uuid}"
CLONE_TEMPLATE = "git clone {url} {{path}}"
def get_name(self):
raise NotImplementedError("Please Implement this method")
def build_remote_url(self):
raise NotImplementedError("Please Implement this method")
def build_remote_delete(self):
return self.REMOTE_DELETE_TEMPLATE.format(uuid=self.get_name())
def build_remote_add(self):
return self.REMOTE_ADD_TEMPLATE.format(
uuid=self.get_name(),
url=self.build_remote_url()
)
def build_clone_cmd(self):
return self.CLONE_TEMPLATE.format(url=self.build_remote_url())
git_cloner/model/repository.py 0 → 100644
from pathlib import Path
import subprocess
import copy
import os
from git_cloner.model.remotes.github_remote import GithubRemote
from git_cloner.utils.config import Config
class Repository(object):
CMD_CHECK = "git -C {} rev-parse"
CMD_INIT = "git init {}"
GITLAB = "gitlab"
GITHUB = "github"
@staticmethod
def clone(remote, root_path):
new_remote = copy.copy(remote)
new_remote.name = "origin"
path = os.path.join(root_path, new_remote.repo_name)
result = subprocess.run(new_remote.build_clone_cmd().format(path=path).split(" "))
if result.returncode != 0:
raise RuntimeError("Error cloning: {}".format(remote.build_remote_url()))
return Repository(path, [new_remote])
def __init__(self, path, remotes=None):
if remotes is None:
self.remotes = {}
else:
self.remotes = remotes
self.path = path
self.name = Path(self.path).name
def add_remote(self, remote):
self.remotes.update({remote.name, remote})
def init_git(self):
subprocess.run(self.CMD_INIT.format(self.path).split(" "))
def check_git(self):
return subprocess.run(self.CMD_CHECK.format(self.path).split(" "),
stdout=subprocess.PIPE,
stderr=subprocess.PIPE).returncode == 0
class RawRepository(object):
def __init__(self, raw_data):
self.raw_data = raw_data
def clone(self, path):
owner = self.raw_data['owner']['login']
repo_name = self.raw_data['name']
remote = GithubRemote(owner, repo_name, Config.get_config().ssh)
return Repository.clone(remote, path)
git_cloner/operations/mass_cloner.py 0 → 100644
from git_cloner.model.owner.owner import Owner
from git_cloner.utils.config import Config
def mass_cloner():
owner = Owner.builder(Config.get_config().type)
out = Config.get_config().dir_out
repos = owner.get_repos()
[repo.clone(out) for repo in repos]
git_cloner/structure_repo.py 0 → 100644
import os
import subprocess
from pathlib import Path
def list_repos(path):
paths = [Path(path).joinpath(e) for e in os.listdir(path)]
return [str(x) for x in paths if check_git_dir(x)]
def check_git_dir(path):
cmd = "git -C {} rev-parse".format(path)
return subprocess.run(cmd.split(" "), stdout=subprocess.PIPE, stderr=subprocess.PIPE).returncode == 0
git_cloner/utils/config.py 0 → 100644
import getpass
import os
# Recursive problems
def read(fname):
with open(os.path.join(os.path.dirname(__file__), fname)) as f:
return f.read().strip()
class Config(object):
_KEY_TARGET_USER = "--target-user"
_KEY_HOST = "--host"
_KEY_TOKEN = "--priv-token"
_KEY_TYPE = "--type"
_KEY_USER = "--user"
_KEY_PASS = "--pass"
_KEY_ORG = "--org"
_KEY_LIMIT = "--limit"
_KEY_DIR_IN = "--in"
_KEY_DIR_OUT = "--out"
_KEY_FORCE = "-f"
_KEY_SAVE = "--save"
_KEY_SSH = "--ssh"
instance = None
def __init__(self, user, password, org, limit, dir_in, dir_out,
force, save, ssh, repo_type, token, host, target_user):
self.target_user = target_user
self.user = user
self.password = password
self.org = org
self.limit = limit
self.dir_in = dir_in
self.dir_out = dir_out
self.force = force
self.save = save
self.ssh = ssh
self.type = repo_type
self.token = token
self.host = host
@classmethod
def load_config(cls, arguments):
user = arguments.pop(cls._KEY_USER, None)
password = arguments.pop(cls._KEY_PASS, None)
org = arguments.pop(cls._KEY_ORG, None)
dir_in = arguments.pop(cls._KEY_DIR_IN, ".")
dir_out = arguments.pop(cls._KEY_DIR_OUT, ".")
force = arguments.pop(cls._KEY_FORCE, False)
save = arguments.pop(cls._KEY_SAVE, None)
ssh = arguments.pop(cls._KEY_SSH, False)
repo_type = arguments.pop(cls._KEY_TYPE, None)
host = arguments.pop(cls._KEY_HOST, None)
path = arguments.pop(cls._KEY_TOKEN, None)
token = read(path) if path is not None else None
target_user = arguments.pop(cls._KEY_TARGET_USER, None)
if user is not None and password is None:
password = getpass.getpass()
cls.instance = Config(user, password, org, 100, dir_in, dir_out,
force, save, ssh, repo_type, token, host, target_user)
return cls.instance
@classmethod
def get_config(cls):
return cls.instance
git_cloner/utils/utils.py 0 → 100644
import os
import requests
from git_cloner.utils.config import Config
def read(fname):
with open(os.path.join(os.path.dirname(__file__), fname)) as f:
return f.read().strip()
def make_request(url, headers=None, auth_if_possible=True):
if headers is None:
headers = {}
config = Config.get_config()
auth = (config.user, config.password) if config.user and auth_if_possible else None
result = requests.get(url, auth=auth, headers=headers)
if result.status_code != 200:
raise RuntimeError("Incorrect request {}\n{}".format(url,str(result.content)))
return result.json()
github_cloner/config.py deleted 100644 → 0
import getpass
class Config(object):
_KEY_USER = "--user"
_KEY_PASS = "--pass"
_KEY_ORG = "--org"
_KEY_LIMIT = "--limit"
_KEY_DIR = "--out"
_KEY_FORCE = "-f"
_KEY_SAVE = "--save"
_KEY_SSH = "--ssh"
# _KEY_OUTPUT =
user = None
password = None
org = None
limit = None
dir = "."
force = False
save = None
ssh = False
limit_per_page = 100
@classmethod
def load_config(cls, arguments):
cls.user = arguments.pop(cls._KEY_USER, None)
cls.password = arguments.pop(cls._KEY_PASS, None)
cls.org = arguments.pop(cls._KEY_ORG, None)
cls.limit = arguments.pop(cls._KEY_LIMIT, None)
cls.dir = arguments.pop(cls._KEY_DIR, ".")
cls.force = arguments.pop(cls._KEY_FORCE, False)
cls.save = arguments.pop(cls._KEY_SAVE, None)
cls.ssh = arguments.pop(cls._KEY_SSH, False)
if cls.user is not None and cls.password is None:
cls.password = getpass.getpass()
return cls
@classmethod
def get_config(cls):
return cls
setup.py
......@@ -16,7 +16,7 @@ setup(name='git_cloner',
author="Francisco Huertas",
author_email="francisco@fhuertas.com",
license="Apache2",
packages=["github_cloner"],
packages=["git_cloner"],
description="Python tool for clon repositories",
long_description=read('README.md'),
url='https://github.com/fhuertas/python_base',
......@@ -27,10 +27,10 @@ setup(name='git_cloner',
"Topic :: Utilities",
"License :: OSI Approved :: Apache Software License",
'Programming Language :: Python :: 3',
'Programming Language :: Python :: 3.5',
'Programming Language :: Python :: 3.6',
],
entry_points={
'console_scripts': [
'git-cloner=github_cloner.boot:main',
'git-cloner=git_cloner.boot:main',
],
},)
tests/git_cloner/model/operations/mass_cloner_it.py 0 → 100644
import json
import unittest
import os
import shutil
import time
from mock_decorators.function_mock import FunctionMock
from git_cloner.model.repository import Repository
from git_cloner.operations import mass_cloner
from git_cloner.utils import utils
from git_cloner.utils.config import Config
import pkg_resources
test_dir = "/tmp/test_tmp"
response_filepath = pkg_resources.resource_filename("tests", "resources/request_fhuertas_org.json")
class MassClonerTest(unittest.TestCase):
def setUp(self):
shutil.rmtree(test_dir, ignore_errors=True)
os.makedirs(test_dir)
time.sleep(1)
def tearDown(self):
shutil.rmtree(test_dir, True)
time.sleep(1)
def serial_it(self):
def test_mass_cloner_user():
config = Config.load_config({})
config.type = "github"
config.target_user = "yunxao"
config.dir_out = test_dir
mass_cloner.mass_cloner()
repos = [name for name in os.listdir(test_dir) if Repository(os.path.join(test_dir, name)).check_git()]
self.assertEqual(len(repos), 10)
def test_mass_cloner_org():
config = Config.load_config({})
config.type = "github"
config.org = "fhuertasorgfortest"
config.ssh = True
config.dir_out = test_dir
mass_cloner.mass_cloner()
repos = [name for name in os.listdir(test_dir) if Repository(os.path.join(test_dir, name)).check_git()]
self.assertEqual(len(repos), 1)
def test_mass_cloner_fail():
config = Config.load_config({})
config.type = "github"
config.org = "fhuertasorgfortest"
config.ssh = True
config.dir_out = test_dir
def mock_request(*args, **kwargs):
result = json.loads(utils.read(response_filepath))
return result
@FunctionMock(utils, 'make_request', mock_request, False)
def inner_test():
mass_cloner.mass_cloner()
inner_test()
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.setUp()
test_mass_cloner_user()
self.setUp()
test_mass_cloner_org()
self.setUp()
test_mass_cloner_fail()
tests/git_cloner/model/owner/github_owner_it.py 0 → 100644
import unittest
from pkg_resources import *
from git_cloner.model.owner.github_owner import GithubOwner
from git_cloner.utils.utils import *
secrets_path = os.path.join(resource_filename('tests', 'resources'), 'secrets')
# class GithubOwnerTest(unittest.TestCase):
class GithubOwnerIT(unittest.TestCase):
def test_get_all_repos_it(self):
config = Config.load_config({})
config.user = read(secrets_path + '/user')
config.password = read(secrets_path + '/password')
repo = read(secrets_path + '/repo')
expected_repo = "{}/{}".format(config.user, repo)
results = GithubOwner(config.user, True).get_repos()
full_names = [repo for repo in results if repo.raw_data.get('full_name') == expected_repo]
self.assertEqual(len(full_names), 1)
tests/git_cloner/model/owner/github_owner_test.py 0 → 100644
import unittest
from mock_decorators.function_mock import FunctionMock
from git_cloner.utils import utils
from git_cloner.model.owner.github_owner import GithubOwner
from git_cloner.utils.config import Config
from tests.git_cloner.utils.gen_utils import *
list_repos = [generate_repo_github() for _ in range(gen_int(101, 501))]
class GithubOwnerTest(unittest.TestCase):
def test_base_api_url_user(self):
owner = GithubOwner("fhuertas", True)
self.assertEqual(owner.api_base_url(), "https://api.github.com/users/fhuertas")
def test_base_api_url_org(self):
owner = GithubOwner("fhuertas", False)
self.assertEqual(owner.api_base_url(), "https://api.github.com/orgs/fhuertas")
def test_get_repos_url(self):
owner = GithubOwner("fhuertas", False)
self.assertEqual(owner.get_repos_url(), "https://api.github.com/orgs/fhuertas/repos?page={p}&per_page=100")
def test_get_all_repos(self):
self.count = 0
Config.load_config({})
def get_repos(*args, **kwargs):
url = args[0]
init = 100 * self.count
end = 100 * (self.count + 1)
self.assertEqual(
url,
"https://api.github.com/users/fhuertas/repos?page={p}&per_page=100".format(p=self.count))
self.count += 1
return list_repos[init:end]
@FunctionMock(utils, 'make_request', get_repos, False)
def inner_test():
owner = GithubOwner("fhuertas", True)
return owner.get_repos()
result = inner_test()
self.assertEqual(len(result), len(list_repos))
def test_get_all_repos_no_result(self):
self.count = 0
Config.load_config({})
def get_repos(*args, **kwargs):
url = args[0]
self.assertEqual(
url,
"https://api.github.com/users/fhuertas/repos?page={p}&per_page=100".format(p=self.count))
self.count += 1
return []
@FunctionMock(utils, 'make_request', get_repos, False)
def inner_test():
owner = GithubOwner("fhuertas", True)
return owner.get_repos()
result = inner_test()
self.assertEqual(len(result), 0)
tests/git_cloner/model/owner/owner_test.py 0 → 100644
import unittest
from git_cloner.model.owner.github_owner import GithubOwner
from git_cloner.model.owner.owner import Owner
from git_cloner.utils.config import Config
class OwnerTest(unittest.TestCase):
def test_builder_github(self):
Config.load_config({}).target_user = "pepe"
result = Owner.builder("github")
self.assertEqual(result.owner, "pepe")
self.assertEqual(result.is_user, True)
self.assertTrue(isinstance(result, GithubOwner))
def test_builder_other(self):
self.assertRaises(RuntimeError, Owner.builder, "invalid")
pass
tests/git_cloner/model/remotes/git_remote_test.py 0 → 100644
import unittest
from git_cloner.model.remotes.github_remote import GithubRemote
class GitRemoteTest(unittest.TestCase):
def test_build_from_url_ssh(self):
remote = GithubRemote.build_from_url("git@github.com:owner/repo-name.git")
self.assertTrue(remote.ssh)
self.assertEqual(remote.owner, "owner")
self.assertEqual(remote.repo_name, "repo-name")
def test_build_from_url_https(self):
remote = GithubRemote.build_from_url("https://github.com/owner/repo-name.git")
self.assertFalse(remote.ssh)
self.assertEqual(remote.owner, "owner")
self.assertEqual(remote.repo_name, "repo-name")
def test_build_remote_url_ssh(self):
remote = GithubRemote("owner", "repo-name", True, "uuid")
self.assertEqual(remote.build_remote_url(), "git@github.com:owner/repo-name.git")
def test_build_remote_url_https(self):
remote = GithubRemote("owner", "repo-name", False, "uuid")
self.assertEqual(remote.build_remote_url(), "https://github.com/owner/repo-name.git")
tests/git_cloner/model/remotes/gitlab_remote_test.py 0 → 100644
import unittest
from git_cloner.model.remotes.gitlab_remote import GitlabRemote
class TestGitlabRemote(unittest.TestCase):
def test_build_remote_ssh(self):
remote = GitlabRemote("git.host.com:2342", "owner", "repo", "ssh")
remote.build_remote_url()
self.assertEqual(remote.build_remote_url(), "ssh://git@git.host.com:2342/owner/repo")
def test_build_remote_http(self):
remote = GitlabRemote("git.host.com:2342", "owner", "repo", "http")
remote.build_remote_url()
self.assertEqual(remote.build_remote_url(), "http://git.host.com:2342/owner/repo")
def test_build_remote_https(self):
remote = GitlabRemote("git.host.com:2342", "owner", "repo", "https")
remote.build_remote_url()
self.assertEqual(remote.build_remote_url(), "https://git.host.com:2342/owner/repo")
tests/git_cloner/model/remotes/remote_test.py 0 → 100644
import unittest
from git_cloner.model.remotes.github_remote import GithubRemote
from git_cloner.model.remotes.gitlab_remote import GitlabRemote
class TestRemote(unittest.TestCase):
def test_clone_cmd(self):
remote = GithubRemote("owner", "repo-name", False, "uuid")
self.assertEqual(remote.build_clone_cmd(), "git clone https://github.com/owner/repo-name.git {path}")
def test_build_remote_add(self):
remote = GitlabRemote("git.host.com:2342", "owner", "repo", "ssh", "uuid")
self.assertEqual(remote.build_remote_add(), "git remote add uuid ssh://git@git.host.com:2342/owner/repo")
def test_build_remote_remove(self):
remote = GitlabRemote("git.host.com:2342", "owner", "repo", "ssh", "uuid")
self.assertEqual(remote.build_remote_delete(), "git remote remove uuid")
tests/git_cloner/model/repository_it.py 0 → 100644
import unittest
import os
import shutil
from git_cloner.model.remotes.github_remote import GithubRemote
from git_cloner.model.repository import Repository, RawRepository
from git_cloner.utils.config import Config
from tests.git_cloner.utils.gen_utils import *
test_dir = "/tmp/test_tmp"
class RepositoryIT(unittest.TestCase):
def setUp(self):
shutil.rmtree(test_dir, ignore_errors=True)
os.makedirs(test_dir)
def tearDown(self):
shutil.rmtree(test_dir, True)
# This tests have problems if are executed in parallel
def serials_it(self):
def creating_with_cloning():
remote = GithubRemote.build_from_url("https://github.com/fhuertas/scala_base.git")
repo = Repository.clone(remote, test_dir)
self.assertTrue(repo.check_git())
self.assertEqual(len(repo.remotes), 1)
self.assertEqual(repo.remotes[0].get_name(), "origin")
self.assertEqual(repo.remotes[0].build_remote_url(), "https://github.com/fhuertas/scala_base.git")
def creating_with_cloning_ssh():
remote = GithubRemote("fhuertas", "mock_decorators", True)
repo = Repository.clone(remote, test_dir)
self.assertTrue(repo.check_git())
self.assertEqual(len(repo.remotes), 1)
self.assertEqual(repo.remotes[0].get_name(), "origin")
self.assertEqual(repo.remotes[0].build_remote_url(), "git@github.com:fhuertas/mock_decorators.git")
def clone_repo_github():
Config.load_config({}).ssh = True
gen_raw_repo = generate_repo_github()
gen_raw_repo['owner']['login'] = "fhuertas"
gen_raw_repo['name'] = "monkey"
raw_repo = RawRepository(gen_raw_repo)
result = raw_repo.clone(test_dir)
self.assertTrue(result.check_git())
creating_with_cloning()
creating_with_cloning_ssh()
clone_repo_github()
tests/git_cloner/model/repository_test.py 0 → 100644
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
tests/github_cloner/test_github_wrapper.py → tests/git_cloner/test_github_wrapper.py
import unittest
from github_cloner import github_wrapper
from github_cloner.config import Config
class TestModule(unittest.TestCase):
def test_run(self):
Config.dir = "pp"
Config.force = False
Config.save = "file.txt"
repos = [{'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': 'cassandra', 'clone_url': 'https://github.com/Stratio/cassandra.git',
'ssh_url': 'git@github.com:Stratio/cassandra.git'}]
github_wrapper.clones(repos)
# Config.dir_out = "pp"
# Config.force = False
# Config.save = "file.txt"
# repos = [{'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': 'cassandra', 'clone_url': 'https://github.com/Stratio/cassandra.git',
# 'ssh_url': 'git@github.com:Stratio/cassandra.git'}]
#
# github_wrapper.clones(repos)
pass
# def test_run_no_dir(self):
# Config.force = False
......
tests/git_cloner/test_gitlab_wrapper.py 0 → 100644
import unittest
from git_cloner.gitlab_wrapper import *
from git_cloner.utils import utils
class TestGitlabWrapper(unittest.TestCase):
def test_push_dir(self):
Config.get_config().org = "org"
Config.get_config().host = "git.labs.com:1234"
Config.get_config().token = utils.read('token')
# push_repository("/tmp/tmp/git_1")
def test_make_all(self):
Config.get_config().org = "stratio"
Config.get_config().host = "git.labs.com:1234"
Config.get_config().token = utils.read('token')
Config.get_config().dir_in = "/home/fhuertas/Projects/personal"
make_all()
tests/github_cloner/test_module.py → tests/git_cloner/test_module.py
import unittest
from github_cloner import boot
from git_cloner import boot
class TestModule(unittest.TestCase):
......
tests/git_cloner/test_structure_repo.py 0 → 100644
import subprocess
import unittest
from pathlib import Path
import os
from pip._vendor.distlib._backport import shutil
from git_cloner import structure_repo
class TestStructureRepo(unittest.TestCase):
_resources_path = "/tmp"
_tmp_path = os.path.join(_resources_path, "tmp")
git = ["git_1", "git_2", "git_3"]
no_git = ["no_git_1", "no_git_2", "no_git_3"]
files = ["file_1", "file_2", "file_3"]
git_paths = []
no_git_paths = []
files_paths = []
def setUp(self):
shutil.rmtree(self._tmp_path, True)
self.git_paths = [os.path.join(self._tmp_path, name) for name in self.git]
self.no_git_paths = [os.path.join(self._tmp_path, name) for name in self.no_git]
self.files_paths = [os.path.join(self._tmp_path, name) for name in self.files]
self.directory = os.getcwd()
os.makedirs(self._tmp_path, mode=0o777, exist_ok=True)
[Path(path).touch() for path in self.files_paths]
[os.makedirs(path, exist_ok=True) for path in self.git_paths]
[os.makedirs(path, exist_ok=True) for path in self.no_git_paths]
cmds = ["git init {}".format(path) for path in self.git_paths]
[subprocess.run(cmd.split(" "), stdout=subprocess.PIPE, check=True) for cmd in cmds]
def tearDown(self):
# shutil.rmtree(self._tmp_path, True)
os.chdir(self.directory)
def test_get_repos(self):
result = set(structure_repo.list_repos(self._tmp_path))
self.assertTrue(result == set(self.git_paths))
def test_check_git_dir_false(self):
[self.assertFalse(structure_repo.check_git_dir(path)) for path in self.no_git_paths]
def test_check_git_dir_true(self):
[self.assertTrue(structure_repo.check_git_dir(path)) for path in self.git_paths]
tests/git_cloner/utils/config_test.py 0 → 100644
import getpass
import unittest
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
class TestConfig(unittest.TestCase):
@FunctionMockResult(entity=getpass, function_name="getpass", result="password", checkExists=False)
@FunctionMockResult(entity=config, function_name="read", result="THIS_IS_A_TOKEN", checkExists=False)
def test_arguments(self):
arguments = {
Config._KEY_USER: "user",
Config._KEY_ORG: "organization",
Config._KEY_LIMIT: "100",
Config._KEY_DIR_OUT: "out",
Config._KEY_FORCE: True,
Config._KEY_SAVE: "save",
Config._KEY_SSH: True,
Config._KEY_DIR_IN: "in",
Config._KEY_TOKEN: "token_path",
Config._KEY_TYPE: "type",
Config._KEY_HOST: "git.host.com",
Config._KEY_TARGET_USER: "pepe"
}
my_config = Config.load_config(arguments)
self.assertEqual(my_config.user, "user")
self.assertEqual(my_config.password, "password")
self.assertEqual(my_config.org, "organization")
self.assertEqual(my_config.limit, 100)
self.assertEqual(my_config.dir_out, "out")
self.assertEqual(my_config.dir_in, "in")
self.assertEqual(my_config.force, True)
self.assertEqual(my_config.save, "save")
self.assertEqual(my_config.ssh, True)
self.assertEqual(my_config.type, "type")
self.assertEqual(my_config.token, "THIS_IS_A_TOKEN")
self.assertEqual(my_config.host, "git.host.com")
self.assertEqual(my_config.target_user, "pepe")
def test_no_arguments(self):
arguments = {}
my_config = Config.load_config(arguments)
self.assertEqual(my_config.user, None)
self.assertEqual(my_config.password, None)
self.assertEqual(my_config.org, None)
self.assertEqual(my_config.limit, 100)
self.assertEqual(my_config.dir_out, ".")
self.assertEqual(my_config.dir_in, ".")
self.assertEqual(my_config.force, False)
self.assertEqual(my_config.save, None)
self.assertEqual(my_config.ssh, False)
self.assertEqual(my_config.type, None)
self.assertEqual(my_config.token, None)
self.assertEqual(my_config.host, None)
self.assertEqual(my_config.target_user, None)
tests/git_cloner/utils/gen_utils.py 0 → 100644
from random import *
names = ['passenger', 'nod', 'pick', 'teeth', 'explain', 'disgusting', 'mix', 'film', 'worry', 'lewd',
'agreeable', 'rub', 'tan', 'suppose', 'waste', 'paddle', 'optimal', 'belief', 'cakes', 'meat',
'shadingo']
def gen_name():
pos = randrange(0, len(names))
return names[pos]
def gen_int(start=0, limit=1000000):
return randrange(start, limit)
def gen_bool():
return bool(randrange(0, 2))
def join_path(path1, path2): return "{}/{}".format(path1, path2)
def gen_github_ssh_url():
pass
def generate_repo_github():
name = gen_name()
user = gen_name()
full_path = join_path(user, name)
description = [gen_name() for _ in range(gen_int(limit=100))]
api_url_base_repos = join_path("https://api.github.com/repos", full_path)
api_user_base = join_path("https://api.github.com/users", user)
repo_base_url = join_path("https://github.com", full_path)
return {
"id": gen_int(),
"name": name,
"full_name": full_path,
"owner": {
"login": user,
"id": gen_int(),
"avatar_url": "https://avatars0.githubusercontent.com/u/5879698?v=4",
"gravatar_id": "",
"url": api_user_base,
"html_url": "https://github.com/{user}".format(user=user),
"followers_url": join_path(api_user_base, "/followers"),
"following_url": join_path(api_user_base, "/following{/other_user}"),
"gists_url": join_path(api_user_base, "/gists{/gist_id}"),
"starred_url": join_path(api_user_base, "/starred{/owner}{/repo}"),
"subscriptions_url": join_path(api_user_base, "/subscriptions"),
"organizations_url": join_path(api_user_base, "/orgs"),
"repos_url": join_path(api_user_base, "/repos"),
"events_url": join_path(api_user_base, "/events{/privacy}"),
"received_events_url": join_path(api_user_base, "/received_events"),
"type": "User",
"site_admin": gen_bool()
},
"private": gen_bool(),
"html_url": repo_base_url,
"description": description,
"fork": gen_bool(),
"url": api_url_base_repos,
"forks_url": join_path(api_url_base_repos, "/forks"),
"keys_url": join_path(api_url_base_repos, "/keys{/key_id}"),
"collaborators_url": join_path(api_url_base_repos, "/collaborators{/collaborator}"),
"teams_url": join_path(api_url_base_repos, "/teams"),
"hooks_url": join_path(api_url_base_repos, "/hooks"),
"issue_events_url": join_path(api_url_base_repos, "/issues/events{/number}"),
"events_url": join_path(api_url_base_repos, "/events"),
"assignees_url": join_path(api_url_base_repos, "/assignees{/user}"),
"branches_url": join_path(api_url_base_repos, "/branches{/branch}"),
"tags_url": join_path(api_url_base_repos, "/tags"),
"blobs_url": join_path(api_url_base_repos, "/git/blobs{/sha}"),
"git_tags_url": join_path(api_url_base_repos, "/git/tags{/sha}"),
"git_refs_url": join_path(api_url_base_repos, "/git/refs{/sha}"),
"trees_url": join_path(api_url_base_repos, "/git/trees{/sha}"),
"statuses_url": join_path(api_url_base_repos, "/statuses/{sha}"),
"languages_url": join_path(api_url_base_repos, "/languages"),
"stargazers_url": join_path(api_url_base_repos, "/stargazers"),
"contributors_url": join_path(api_url_base_repos, "/contributors"),
"subscribers_url": join_path(api_url_base_repos, "/subscribers"),
"subscription_url": join_path(api_url_base_repos, "/subscription"),
"commits_url": join_path(api_url_base_repos, "/commits{/sha}"),
"git_commits_url": join_path(api_url_base_repos, "/git/commits{/sha}"),
"comments_url": join_path(api_url_base_repos, "/comments{/number}"),
"issue_comment_url": join_path(api_url_base_repos, "/issues/comments{/number}"),
"contents_url": join_path(api_url_base_repos, "/contents/{+path}"),
"compare_url": join_path(api_url_base_repos, "/compare/{base}...{head}"),
"merges_url": join_path(api_url_base_repos, "/merges"),
"archive_url": join_path(api_url_base_repos, "/{archive_format}{/ref}"),
"downloads_url": join_path(api_url_base_repos, "/downloads"),
"issues_url": join_path(api_url_base_repos, "/issues{/number}"),
"pulls_url": join_path(api_url_base_repos, "/pulls{/number}"),
"milestones_url": join_path(api_url_base_repos, "/milestones{/number}"),
"notifications_url": join_path(api_url_base_repos, "/notifications{?since,all,participating}"),
"labels_url": join_path(api_url_base_repos, "/labels{/name}"),
"releases_url": join_path(api_url_base_repos, "/releases{/id}"),
"deployments_url": join_path(api_url_base_repos, "/deployments"),
"created_at": "2017-05-04T06:30:53Z",
"updated_at": "2017-05-05T14:02:57Z",
"pushed_at": "2017-05-05T14:02:56Z",
"git_url": "git://github.com/{}.git".format(full_path),
"ssh_url": "git@github.com:{}.git".format(full_path),
"clone_url": join_path(repo_base_url, ".git"),
"svn_url": repo_base_url,
"homepage": None,
"size": gen_int(),
"stargazers_count": 0,
"watchers_count": 0,
"language": "Shell",
"has_issues": gen_bool(),
"has_projects": gen_bool(),
"has_downloads": gen_bool(),
"has_wiki": gen_bool(),
"has_pages": gen_bool(),
"forks_count": 0,
"mirror_url": None,
"archived": gen_bool(),
"open_issues_count": 0,
"forks": 0,
"open_issues": 0,
"watchers": 0,
"default_branch": "master"
}
tests/git_cloner/utils/utils_it.py 0 → 100644
import unittest
from pkg_resources import *
from git_cloner.utils.utils import *
secrets_path = os.path.join(resource_filename('tests', 'resources'), 'secrets')
class UtilsTest(unittest.TestCase):
def make_curl_with_auth_incorrect_it(self):
auth_url = "https://api.github.com"
config = Config.load_config({})
config.user = "invalid"
config.password = "invalid"
self.assertRaises(RuntimeError, make_request, auth_url)
def make_curl_with_auth_valid_it(self):
auth_url = "https://api.github.com"
config = Config.load_config({})
config.user = read(secrets_path + '/user')
config.password = read(secrets_path + '/password')
result = make_request(auth_url)
self.assertEqual(result.get('current_user_url'), "https://api.github.com/user")
def make_curl_with_no_auth_it(self):
auth_url = "https://api.github.com"
Config.load_config({})
result = make_request(auth_url)
self.assertEqual(result.get('current_user_url'), "https://api.github.com/user")
tests/github_cloner/file.txt deleted 100644 → 0
* dg-agent-commons CANNOT be cloned
* cassandra CANNOT be cloned
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