Add Repository and Package classes
This commit is contained in:
121
blenderpack.py
121
blenderpack.py
@@ -12,6 +12,7 @@ import ast
|
||||
import argparse
|
||||
import zipfile
|
||||
import logging
|
||||
from collections import MutableMapping
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
@@ -19,38 +20,114 @@ REQUIRED_KEYS = set(['name', 'blender', 'version'])
|
||||
SCHEMA_VERSION = 1
|
||||
|
||||
class BadAddon(Exception):
|
||||
pass
|
||||
|
||||
|
||||
# TODO add repolist class
|
||||
|
||||
def load_repo(url):
|
||||
"""
|
||||
Searches local repo.json for repository with a matching 'url' and returns it
|
||||
Raised when something expected to be an addon turns out not to be
|
||||
"""
|
||||
pass #TODO
|
||||
|
||||
def write_repo(repo):
|
||||
class RepositoryListError(ValueError):
|
||||
"""
|
||||
Writes repo to local repolist json
|
||||
Raised when something is amiss with the repo.json file
|
||||
"""
|
||||
pass #TODO
|
||||
|
||||
def fetch_repo(url: str) -> dict:
|
||||
""" Requests repo.json from URL and embeds etag/last-modification headers"""
|
||||
local_copy = load_repo(url)
|
||||
headers = {}
|
||||
if 'etag' in local_copy: headers['If-None-Match'] = local_copy['etag']
|
||||
if 'last-modified' in local_copy: headers['If-Modified-Since'] = local_copy['last-modified']
|
||||
class RepositoryNotFoundError(RepositoryListError):
|
||||
"""
|
||||
Raised when looking for a repository which isn't there
|
||||
"""
|
||||
|
||||
resp = requests.get(url, headers=headers)
|
||||
class Package:
|
||||
"""
|
||||
Stores package methods and metadata
|
||||
"""
|
||||
|
||||
repo = json.loads(resp.json())
|
||||
repo['etag'] = resp.headers.get('etag')
|
||||
repo['last-modified'] = resp.headers.get('last-modified')
|
||||
def __init__(self, package_dict:dict = None):
|
||||
self.from_dict(package_dict)
|
||||
|
||||
write_repo(repo)
|
||||
def to_dict(self) -> dict:
|
||||
"""
|
||||
Return a dict representation of the package
|
||||
"""
|
||||
return {
|
||||
'bl_info': self.bl_info,
|
||||
'url': self.url,
|
||||
}
|
||||
|
||||
def from_dict(self, package_dict: dict):
|
||||
"""
|
||||
Get attributes from a dict such as produced by `to_dict`
|
||||
"""
|
||||
if package_dict is None:
|
||||
package_dict = {}
|
||||
|
||||
for attr in ('name', 'url', 'bl_info'):
|
||||
setattr(self, attr, package_dict.get(attr))
|
||||
|
||||
|
||||
class Repository:
|
||||
"""
|
||||
Stores repository metadata (including packages)
|
||||
"""
|
||||
|
||||
def __init__(self, repo_dict:dict = None):
|
||||
self.from_dict(repo_dict)
|
||||
|
||||
def refresh(self):
|
||||
"""
|
||||
Requests repo.json from URL and embeds etag/last-modification headers
|
||||
"""
|
||||
|
||||
req_headers = {}
|
||||
|
||||
# Do things this way to avoid adding empty objects/None to the req_headers dict
|
||||
if 'etag' in self._headers:
|
||||
req_headers['If-None-Match'] = self._headers['etag']
|
||||
if 'last-modified' in self._headers:
|
||||
req_headers['If-Modified-Since'] = self._headers['last-modified']
|
||||
|
||||
#try
|
||||
resp = requests.get(self.url, headers=req_headers)
|
||||
|
||||
repodict = json.loads(resp.json())
|
||||
repodict['etag'] = resp.headers.get('etag')
|
||||
repodict['last-modified'] = resp.headers.get('last-modified')
|
||||
|
||||
self.from_dict(repodict)
|
||||
|
||||
|
||||
def to_dict(self) -> dict:
|
||||
"""
|
||||
Return a dict representation of the repository
|
||||
"""
|
||||
return {
|
||||
'name': self.name,
|
||||
'packages': [p.to_dict() for p in self.packages],
|
||||
'url': self.url,
|
||||
'_headers': self._headers,
|
||||
}
|
||||
|
||||
def from_dict(self, repodict: dict):
|
||||
"""
|
||||
Get attributes from a dict such as produced by `to_dict`
|
||||
"""
|
||||
if repodict is None:
|
||||
repodict = {}
|
||||
|
||||
for attr in ('name', 'url', 'packages', '_headers'):
|
||||
if attr == 'package':
|
||||
value = set(Package(pkg) for pkg in repodict.get('packages', []))
|
||||
else:
|
||||
value = repodict.get(attr)
|
||||
|
||||
setattr(self, attr, value)
|
||||
|
||||
def dump(self, path: pathlib.Path):
|
||||
"""
|
||||
Dump repository as a repo.json file in 'path'
|
||||
"""
|
||||
with (path / 'repo.json').open('w', encoding='utf-8') as repo_file:
|
||||
json.dump(self.to_dict(), repo_file, indent=4, sort_keys=True)
|
||||
log.info("repo.json written to %s" % path)
|
||||
|
||||
# def fetch_repo(url: str, repo: Repository) -> dict:
|
||||
def refresh(url):
|
||||
# we have to explicitly close the end of the pipe we are NOT using,
|
||||
# otherwise no exception will be generated when the other process closes its end.
|
||||
|
65
tests/test_repo_io.py
Executable file
65
tests/test_repo_io.py
Executable file
@@ -0,0 +1,65 @@
|
||||
#!/usr/bin/env python3
|
||||
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
import logging
|
||||
import json
|
||||
import blenderpack as BP
|
||||
|
||||
logging.basicConfig(level=logging.DEBUG,
|
||||
format='%(levelname)8s: %(message)s')
|
||||
|
||||
class TestRepoInstantiation(unittest.TestCase):
|
||||
"""
|
||||
Tests of the creation of a Repository object
|
||||
"""
|
||||
|
||||
# helper_path = Path('tests/test_helpers')
|
||||
# repos = blenderpack.Repositories(helper_path / 'repo.json')
|
||||
|
||||
# def test_load(self):
|
||||
# repo = self.repos.load('http://someurl.tld/repo.json')
|
||||
|
||||
repo_dict = {
|
||||
'name': 'The Best Repo Ever',
|
||||
'url': 'http://someurl.tld/repo.json',
|
||||
'packages': [
|
||||
{'name': 'pkg1'},
|
||||
{'name': 'pkg2'},
|
||||
],
|
||||
}
|
||||
|
||||
def test_create_from_dict(self):
|
||||
"""
|
||||
Instantiate repository repository with a dict and check
|
||||
if all the items are carried over
|
||||
"""
|
||||
repodict = self.repo_dict
|
||||
repo = BP.Repository(repodict)
|
||||
for key, val in repodict.items():
|
||||
self.assertEqual(getattr(repo, key), val)
|
||||
|
||||
def test_create_from_none(self):
|
||||
"""
|
||||
Instantiate repository repository from none and check that
|
||||
the new repository's properties are set to none
|
||||
"""
|
||||
repodict = self.repo_dict
|
||||
repo = BP.Repository(None)
|
||||
for key, val in repodict.items():
|
||||
self.assertEqual(getattr(repo, key), None)
|
||||
|
||||
def test_create_from_incomplete(self):
|
||||
"""
|
||||
Instantiate repository repository from a partial dict
|
||||
and check that all properties are set, either to None or to the
|
||||
value from the dict
|
||||
"""
|
||||
repodict = {
|
||||
'name': 'The Best Repo Ever',
|
||||
}
|
||||
repo = BP.Repository(repodict)
|
||||
for key, val in repodict.items():
|
||||
self.assertEqual(getattr(repo, key), val)
|
||||
self.assertIs(repo.url, None)
|
||||
|
Reference in New Issue
Block a user