Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
class ConfigFactory(PublishFactory):
path = pathlib.Path(__file__).parent / 'configs' / 'basic.toml'
class Meta:
model = config_module.Config
class RepoFactory(PublishFactory):
config = factory.SubFactory(ConfigFactory)
name = factory.Faker('slug')
git_repo_url = factory.Faker('url')
secret = factory.Faker('pystr', min_chars=20, max_chars=20)
class Meta:
model = publishing.GenericRepo
class GithubRepoFactory(RepoFactory):
class Meta:
model = publishing.GithubRepo
def test_publish_repo_bins_fails(self, mocker):
mocker.patch.object(git.Repo, 'clone_from')
mocker.patch.object(shutil, 'rmtree')
ipfs_client_mock = mocker.Mock(spec=ipfshttpclient.Client)
ipfs_client_mock.add.return_value = [{'Hash': 'some-hash'}]
mocker.patch.object(ipfshttpclient, 'connect')
ipfshttpclient.connect.return_value = ipfs_client_mock
mocker.patch.object(subprocess, 'run')
subprocess.run.return_value = subprocess.CompletedProcess(None, 1)
repo: publishing.GenericRepo = factories.RepoFactory(build_bin='some_cmd', after_publish_bin='some_other_cmd')
with pytest.raises(exceptions.RepoException):
repo.publish_repo()
def publish(ctx, name):
"""
Will immediately publish repo based on its configuration.
"""
config: config_module.Config = ctx.obj['config']
repo: publishing.GenericRepo = config.repos.get(name)
if repo is None:
click.secho('Unknown repo!', fg='red')
exit(1)
repo.publish_repo()
config.save()
click.echo('Repo successfully published!')
def handler_dispatcher(repo: typing.Union[publishing.GenericRepo, publishing.GithubRepo]) -> 'GenericHandler':
"""
Dispatch request to proper Handler based on what kind of repo the request is directed to.
:param repo: Name of the repo
:return:
"""
if type(repo) is publishing.GenericRepo:
return GenericHandler(repo)
elif type(repo) is publishing.GithubRepo:
return GithubHandler(repo)
else:
raise exceptions.HttpException('Unknown Repo\'s class!')
def _load_data(self,
data): # type: (typing.Dict[str, typing.Any]) -> typing.Tuple[dict, typing.Dict[str, publishing.Repo]]
from publish import publishing
self._verify_data(data)
repos: typing.Dict[str, publishing.GenericRepo] = {}
for value in data.pop('repos', {}).values():
repo_class = publishing.get_repo_class(value['git_repo_url'])
repo = repo_class.from_toml_dict(value, self)
repos[repo.name] = repo
return data, repos
continue
ipns_addr = f'/ipns/{out["Id"]}/'
else:
keys = config.ipfs.key.list()
key_object = next((x for x in keys['Keys'] if x['Name'] == ipns_key), None)
if key_object is None:
logger.info('The passed IPNS key name \'{}\' was not found, generating new key with this name')
key_object = config.ipfs.key.gen(ipns_key, IPNS_KEYS_TYPE)
ipns_addr = f'/ipns/{key_object["Id"]}/'
return ipns_key, ipns_addr
class GithubRepo(GenericRepo):
"""
Special case of Repo specific to GitHub hosted repos.
"""
def __init__(self, git_repo_url, **kwargs):
if not is_github_url(git_repo_url):
raise exceptions.RepoException('The passed Git repo URL is not related to GitHub!')
super().__init__(git_repo_url=git_repo_url, **kwargs)
@property
def webhook_url(self):
return f'{self.config.webhook_base}/publish/{self.name}'
def bootstrap_repo(config: config_module.Config, git_repo_url=None, **kwargs) -> repo_instance:
"""
Initiate the interactive bootstrap process of creating new Repo's instance
:param config:
:param git_repo_url:
:param kwargs:
:return:
"""
if git_repo_url is None:
git_repo_url = inquirer.shortcuts.text('Git URL of the repo', validate=lambda _, x: validate_repo(x))
if is_github_url(git_repo_url):
return GithubRepo.bootstrap_repo(config, git_repo_url=git_repo_url, **kwargs)
return GenericRepo.bootstrap_repo(config, git_repo_url=git_repo_url, **kwargs)
def get_repo_class(url: str) -> repo_class:
"""
For Git repo's URL it returns appropriate class that should represents the repo.
:param url:
:return:
"""
if is_github_url(url):
return GithubRepo
return GenericRepo