gronod/headphones
1
0
Fork 1
mirror of https://github.com/rembo10/headphones.git synced 2026-05-16 08:35:32 +01:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'develop'. Includes updated github api,

Browse Source 
refreshing artists based on when they were last
updated, include releases with no release date,
hardcoded size limits removed
This commit is contained in:
rembo10
2012-06-18 16:35:15 +05:30
parent ddf3432f75 d5b8a90ae4
commit 89c2a8c3ba
11 changed files with 74 additions and 1255 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
headphones/__init__.py
View File

@@ -657,7 +657,7 @@ def dbcheck():
conn=sqlite3.connect(DB_FILE)
c=conn.cursor()
c.execute('CREATE TABLE IF NOT EXISTS artists (ArtistID TEXT UNIQUE, ArtistName TEXT, ArtistSortName TEXT, DateAdded TEXT, Status TEXT, IncludeExtras INTEGER, LatestAlbum TEXT, ReleaseDate TEXT, AlbumID TEXT, HaveTracks INTEGER, TotalTracks INTEGER)')
c.execute('CREATE TABLE IF NOT EXISTS artists (ArtistID TEXT UNIQUE, ArtistName TEXT, ArtistSortName TEXT, DateAdded TEXT, Status TEXT, IncludeExtras INTEGER, LatestAlbum TEXT, ReleaseDate TEXT, AlbumID TEXT, HaveTracks INTEGER, TotalTracks INTEGER, LastUpdated TEXT)')
c.execute('CREATE TABLE IF NOT EXISTS albums (ArtistID TEXT, ArtistName TEXT, AlbumTitle TEXT, AlbumASIN TEXT, ReleaseDate TEXT, DateAdded TEXT, AlbumID TEXT UNIQUE, Status TEXT, Type TEXT)')
c.execute('CREATE TABLE IF NOT EXISTS tracks (ArtistID TEXT, ArtistName TEXT, AlbumTitle TEXT, AlbumASIN TEXT, AlbumID TEXT, TrackTitle TEXT, TrackDuration, TrackID TEXT, TrackNumber INTEGER, Location TEXT, BitRate INTEGER, CleanName TEXT, Format TEXT)')
c.execute('CREATE TABLE IF NOT EXISTS snatched (AlbumID TEXT, Title TEXT, Size INTEGER, URL TEXT, DateAdded TEXT, Status TEXT, FolderName TEXT)')
@@ -747,7 +747,12 @@ def dbcheck():
try:
c.execute('SELECT Format from tracks')
except sqlite3.OperationalError:
c.execute('ALTER TABLE tracks ADD COLUMN Format TEXT DEFAULT NULL')
c.execute('ALTER TABLE tracks ADD COLUMN Format TEXT DEFAULT NULL')
try:
c.execute('SELECT LastUpdated from artists')
except sqlite3.OperationalError:
c.execute('ALTER TABLE artists ADD COLUMN LastUpdated TEXT DEFAULT NULL')
conn.commit()
c.close()

2
headphones/importer.py
View File

@@ -222,6 +222,8 @@ def addArtisttoDB(artistid, extrasonly=False):
newValueDict = {"Status": "Active",
"TotalTracks": totaltracks,
"HaveTracks": havetracks}
newValueDict['LastUpdated'] = helpers.now()
myDB.upsert("artists", newValueDict, controlValueDict)
logger.info(u"Updating complete for: " + artist['artist_name'])

16
headphones/mb.py
View File

@@ -285,14 +285,7 @@ def getReleaseGroup(rgid):
time.sleep(5)
if not releaseResult:
continue
#skip releases that do not have any release date associated with them
#if we don't do this the albums will get the release date "None" and will
#automatically be set to wanted if "Automatically Mark Upcoming Albums as Wanted" is set
#because None is not a valid date and will never be in the past so it is always upcoming
if 'date' not in releaseResult:
continue
continue
if releaseGroup['type'] == 'live' and releaseResult['status'] != 'Official':
logger.debug('%s is not an official live album. Skipping' % releaseResult.name)
@@ -342,7 +335,7 @@ def getReleaseGroup(rgid):
'asin': unicode(releaseResult.get('asin')) if 'asin' in releaseResult else None,
'trackscount': totalTracks,
'releaseid': unicode(releaseResult.get('id')),
'releasedate': unicode(releaseResult.get('date')),
'releasedate': unicode(releaseResult.get('date')) if 'date' in releaseResult else None,
'format': format,
'country': country
}
@@ -350,6 +343,9 @@ def getReleaseGroup(rgid):
releaselist.append(release_dict)
#necessary to make dates that miss the month and/or day show up after full dates
def getSortableReleaseDate(releaseDate):
if releaseDate == None:
return 'None';#change this value to change the sorting behaviour of none, returning 'None' will put it at the top
#which was normal behaviour for pre-ngs versions
if releaseDate.count('-') == 2:
return releaseDate
elif releaseDate.count('-') == 1:
@@ -366,7 +362,7 @@ def getReleaseGroup(rgid):
a = multikeysort(releaselist, ['-hasasin', 'country', 'format', 'trackscount_delta'])
release_dict = {'releaseid' :a[0]['releaseid'],
'releasedate' : unicode(releaselist[0]['releasedate']),
'releasedate' : releaselist[0]['releasedate'],
'trackcount' : a[0]['trackscount'],
'tracks' : a[0]['tracks'],
'asin' : a[0]['asin'],

52
headphones/searcher.py
View File

@@ -141,14 +141,11 @@ def searchNZB(albumid=None, new=False, losslessOnly=False):
if headphones.NZBMATRIX:
provider = "nzbmatrix"
if headphones.PREFERRED_QUALITY == 3 or losslessOnly:
categories = "23"
maxsize = 10000000000
categories = "23"
elif headphones.PREFERRED_QUALITY:
categories = "23,22"
maxsize = 2000000000
else:
categories = "22"
maxsize = 300000000
# For some reason NZBMatrix is erroring out/timing out when the term starts with a "The" right now
# so we'll strip it out for the time being. This may get fixed on their end, it may not, but
@@ -186,12 +183,10 @@ def searchNZB(albumid=None, new=False, losslessOnly=False):
url = item.link
title = item.title
size = int(item.links[1]['length'])
if size < maxsize:
resultlist.append((title, size, url, provider))
logger.info('Found %s. Size: %s' % (title, helpers.bytes_to_mb(size)))
else:
logger.info('%s is larger than the maxsize for this category, skipping. (Size: %i bytes)' % (title, size))
resultlist.append((title, size, url, provider))
logger.info('Found %s. Size: %s' % (title, helpers.bytes_to_mb(size)))
except AttributeError, e:
logger.info(u"No results found from NZBMatrix for %s" % term)
@@ -199,13 +194,10 @@ def searchNZB(albumid=None, new=False, losslessOnly=False):
provider = "newznab"
if headphones.PREFERRED_QUALITY == 3 or losslessOnly:
categories = "3040"
maxsize = 10000000000
elif headphones.PREFERRED_QUALITY:
categories = "3040,3010"
maxsize = 2000000000
else:
categories = "3010"
maxsize = 300000000
categories = "3010"
params = { "t": "search",
"apikey": headphones.NEWZNAB_APIKEY,
@@ -238,11 +230,9 @@ def searchNZB(albumid=None, new=False, losslessOnly=False):
url = item.link
title = item.title
size = int(item.links[1]['length'])
if size < maxsize:
resultlist.append((title, size, url, provider))
logger.info('Found %s. Size: %s' % (title, helpers.bytes_to_mb(size)))
else:
logger.info('%s is larger than the maxsize for this category, skipping. (Size: %i bytes)' % (title, size))
resultlist.append((title, size, url, provider))
logger.info('Found %s. Size: %s' % (title, helpers.bytes_to_mb(size)))
except Exception, e:
logger.error(u"An unknown error occured trying to parse the feed: %s" % e)
@@ -251,13 +241,10 @@ def searchNZB(albumid=None, new=False, losslessOnly=False):
provider = "nzbsorg"
if headphones.PREFERRED_QUALITY == 3 or losslessOnly:
categories = "3040"
maxsize = 10000000000
elif headphones.PREFERRED_QUALITY:
categories = "3040,3010"
maxsize = 2000000000
else:
categories = "3010"
maxsize = 300000000
categories = "3010"
params = { "t": "search",
"apikey": headphones.NZBSORG_HASH,
@@ -290,12 +277,10 @@ def searchNZB(albumid=None, new=False, losslessOnly=False):
url = item.link
title = item.title
size = int(item.links[1]['length'])
if size < maxsize:
resultlist.append((title, size, url, provider))
logger.info('Found %s. Size: %s' % (title, helpers.bytes_to_mb(size)))
else:
logger.info('%s is larger than the maxsize for this category, skipping. (Size: %i bytes)' % (title, size))
resultlist.append((title, size, url, provider))
logger.info('Found %s. Size: %s' % (title, helpers.bytes_to_mb(size)))
except Exception, e:
logger.error(u"An unknown error occured trying to parse the feed: %s" % e)
@@ -305,15 +290,12 @@ def searchNZB(albumid=None, new=False, losslessOnly=False):
if headphones.PREFERRED_QUALITY == 3 or losslessOnly:
categories = "7" #music
format = "2" #flac
maxsize = 10000000000
elif headphones.PREFERRED_QUALITY:
categories = "7" #music
format = "10" #mp3+flac
maxsize = 2000000000
else:
categories = "7" #music
format = "8" #mp3
maxsize = 300000000
format = "8" #mp3
params = {
"fpn": "p",
@@ -370,11 +352,11 @@ def searchNZB(albumid=None, new=False, losslessOnly=False):
logger.info("Didn't find a valid Newzbin reportid in linknode")
else:
url = id_match.group(1) #we have to make a post request later, need the id
if size < maxsize and url:
if url:
resultlist.append((title, size, url, provider))
logger.info('Found %s. Size: %s' % (title, helpers.bytes_to_mb(size)))
else:
logger.info('%s is larger than the maxsize for this category, skipping. (Size: %i bytes)' % (title, size))
logger.info('No url link found in nzb. Skipping.')
else:
logger.info('No results found from NEWZBIN for %s' % term)

2
headphones/updater.py
View File

@@ -6,7 +6,7 @@ def dbUpdate():
myDB = db.DBConnection()
activeartists = myDB.select('SELECT ArtistID, ArtistName from artists WHERE Status="Active" or Status="Loading" order by ArtistSortName collate nocase')
activeartists = myDB.select('SELECT ArtistID, ArtistName from artists WHERE Status="Active" or Status="Loading" order by LastUpdated ASC')
logger.info('Starting update for %i active artists' % len(activeartists))

72
headphones/versioncheck.py
View File

@@ -1,13 +1,12 @@
import platform, subprocess, re, os
import urllib2
import tarfile
import platform, subprocess, re, os, urllib2, tarfile
import headphones
from headphones import logger, version
from lib.pygithub import github
import lib.simplejson as simplejson
user = "rembo10"
branch = "master"
def runGit(args):
@@ -92,33 +91,44 @@ def getVersion():
def checkGithub():
commits_behind = 0
cur_commit = headphones.CURRENT_VERSION
latest_commit = None
gh = github.GitHub()
# Get the latest commit available from github
url = 'https://api.github.com/repos/%s/headphones/commits/%s' % (user, branch)
logger.info ('Retrieving latest version information from github')
try:
result = urllib2.urlopen(url).read()
git = simplejson.JSONDecoder().decode(result)
headphones.LATEST_VERSION = git['sha']
except:
logger.warn('Could not get the latest commit from github')
headphones.COMMITS_BEHIND = 0
return headphones.CURRENT_VERSION
for curCommit in gh.commits.forBranch('rembo10', 'headphones', version.HEADPHONES_VERSION):
if not latest_commit:
latest_commit = curCommit.id
if not cur_commit:
break
# See how many commits behind we are
if headphones.CURRENT_VERSION:
logger.info('Comparing currently installed version with latest github version')
url = 'https://api.github.com/repos/%s/headphones/compare/%s...%s' % (user, headphones.CURRENT_VERSION, headphones.LATEST_VERSION)
if curCommit.id == cur_commit:
break
try:
result = urllib2.urlopen(url).read()
git = simplejson.JSONDecoder().decode(result)
headphones.COMMITS_BEHIND = git['total_commits']
except:
logger.warn('Could not get commits behind from github')
headphones.COMMITS_BEHIND = 0
return headphones.CURRENT_VERSION
commits_behind += 1
headphones.LATEST_VERSION = latest_commit
headphones.COMMITS_BEHIND = commits_behind
if headphones.LATEST_VERSION == headphones.CURRENT_VERSION:
logger.info('Headphones is already up-to-date.')
return latest_commit
if headphones.COMMITS_BEHIND >= 1:
logger.info('New version is available. You are %s commits behind' % headphones.COMMITS_BEHIND)
elif headphones.COMMITS_BEHIND == 0:
logger.info('Headphones is up to date')
elif headphones.COMMITS_BEHIND == -1:
logger.info('You are running an unknown version of Headphones. Run the updater to identify your version')
else:
logger.info('You are running an unknown version of Headphones. Run the updater to identify your version')
return headphones.LATEST_VERSION
def update():
@@ -146,7 +156,7 @@ def update():
else:
tar_download_url = 'http://github.com/rembo10/headphones/tarball/'+version.HEADPHONES_VERSION
tar_download_url = 'https://github.com/%s/headphones/tarball/%s' % (user, branch)
update_dir = os.path.join(headphones.PROG_DIR, 'update')
version_path = os.path.join(headphones.PROG_DIR, 'version.txt')
@@ -200,5 +210,5 @@ def update():
ver_file.write(headphones.LATEST_VERSION)
ver_file.close()
except IOError, e:
logger.error(u"Unable to write for curCommit in gh.commits.forBranch file, update not complete: "+ex(e))
return
logger.error(u"Unable to write current version to version.txt, update not complete: "+ex(e))
return

25
lib/pygithub/__init__.py
View File

@@ -1,25 +0,0 @@
# Copyright (c) 2005-2008 Dustin Sallings <dustin@spy.net>
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.
#
# <http://www.opensource.org/licenses/mit-license.php>
"""
github module.
"""
__all__ = ['github','ghsearch','githubsync']

50
lib/pygithub/ghsearch.py
View File

@@ -1,50 +0,0 @@
#!/usr/bin/env python
#
# Copyright (c) 2005-2008 Dustin Sallings <dustin@spy.net>
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.
#
# <http://www.opensource.org/licenses/mit-license.php>
"""
Search script.
"""
import sys
import github
def usage():
"""display the usage and exit"""
print "Usage: %s keyword [keyword...]" % (sys.argv[0])
sys.exit(1)
def mk_url(repo):
return "http://github.com/%s/%s" % (repo.username, repo.name)
if __name__ == '__main__':
g = github.GitHub()
if len(sys.argv) < 2:
usage()
res = g.repos.search(' '.join(sys.argv[1:]))
for repo in res:
try:
print "Found %s at %s" % (repo.name, mk_url(repo))
except AttributeError:
print "Bug: Couldn't format %s" % repo.__dict__

520
lib/pygithub/github.py
View File

@@ -1,520 +0,0 @@
#!/usr/bin/env python
#
# Copyright (c) 2005-2008 Dustin Sallings <dustin@spy.net>
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.
#
# <http://www.opensource.org/licenses/mit-license.php>
"""
Interface to github's API (v2).
Basic usage:
g = GitHub()
for r in g.user.search('dustin'):
print r.name
See the GitHub docs or README.markdown for more usage.
Copyright (c) 2007 Dustin Sallings <dustin@spy.net>
"""
# GAE friendly URL detection (theoretically)
try:
import urllib2
default_fetcher = urllib2.urlopen
except LoadError:
pass
import urllib
import xml
import xml.dom.minidom
def _string_parser(x):
"""Extract the data from the first child of the input."""
return x.firstChild.data
_types = {
'string': _string_parser,
'integer': lambda x: int(_string_parser(x)),
'float': lambda x: float(_string_parser(x)),
'datetime': _string_parser,
'boolean': lambda x: _string_parser(x) == 'true'
}
def _parse(el):
"""Generic response parser."""
type = 'string'
if el.attributes and 'type' in el.attributes.keys():
type = el.attributes['type'].value
elif el.localName in _types:
type = el.localName
elif len(el.childNodes) > 1:
# This is a container, find the child type
type = None
ch = el.firstChild
while ch and not type:
if ch.localName == 'type':
type = ch.firstChild.data
ch = ch.nextSibling
if not type:
raise Exception("Can't parse %s, known: %s"
% (el.toxml(), repr(_types.keys())))
return _types[type](el)
def parses(t):
"""Parser for a specific type in the github response."""
def f(orig):
orig.parses = t
return orig
return f
def with_temporary_mappings(m):
"""Allow temporary localized altering of type mappings."""
def f(orig):
def every(self, *args):
global _types
o = _types.copy()
for k,v in m.items():
if v:
_types[k] = v
else:
del _types[k]
try:
return orig(self, *args)
finally:
_types = o
return every
return f
@parses('array')
def _parseArray(el):
rv = []
ch = el.firstChild
while ch:
if ch.nodeType != xml.dom.Node.TEXT_NODE and ch.firstChild:
rv.append(_parse(ch))
ch=ch.nextSibling
return rv
class BaseResponse(object):
"""Base class for XML Response Handling."""
def __init__(self, el):
ch = el.firstChild
while ch:
if ch.nodeType != xml.dom.Node.TEXT_NODE and ch.firstChild:
ln = ch.localName.replace('-', '_')
self.__dict__[ln] = _parse(ch)
ch=ch.nextSibling
def __repr__(self):
return "<<%s>>" % str(self.__class__)
class User(BaseResponse):
"""A github user."""
parses = 'user'
def __repr__(self):
return "<<User %s>>" % self.name
class Plan(BaseResponse):
"""A github plan."""
parses = 'plan'
def __repr__(self):
return "<<Plan %s>>" % self.name
class Repository(BaseResponse):
"""A repository."""
parses = 'repository'
@property
def owner_name(self):
if hasattr(self, 'owner'):
return self.owner
else:
return self.username
def __repr__(self):
return "<<Repository %s/%s>>" % (self.owner_name, self.name)
class PublicKey(BaseResponse):
"""A public key."""
parses = 'public-key'
title = 'untitled'
def __repr__(self):
return "<<Public key %s>>" % self.title
class Commit(BaseResponse):
"""A commit."""
parses = 'commit'
def __repr__(self):
return "<<Commit: %s>>" % self.id
class Parent(Commit):
"""A commit parent."""
parses = 'parent'
class Author(User):
"""A commit author."""
parses = 'author'
class Committer(User):
"""A commit committer."""
parses = 'committer'
class Issue(BaseResponse):
"""An issue within the issue tracker."""
parses = 'issue'
def __repr__(self):
return "<<Issue #%d>>" % self.number
class Label(BaseResponse):
"""A Label within the issue tracker."""
parses = 'label'
def __repr__(self):
return "<<Label $%d>>" % self.number
class Tree(BaseResponse):
"""A Tree object."""
# Parsing is scoped to objects...
def __repr__(self):
return "<<Tree: %s>>" % self.name
class Blob(BaseResponse):
"""A Blob object."""
# Parsing is scoped to objects...
def __repr__(self):
return "<<Blob: %s>>" % self.name
class Modification(BaseResponse):
"""A modification object."""
# Parsing is scoped to usage
def __repr__(self):
return "<<Modification of %s>>" % self.filename
class Network(BaseResponse):
"""A network entry."""
parses = 'network'
def __repr__(self):
return "<<Network of %s/%s>>" % (self.owner, self.name)
# Load the known types.
for __t in (t for t in globals().values() if hasattr(t, 'parses')):
_types[__t.parses] = __t
class BaseEndpoint(object):
BASE_URL = 'http://github.com/api/v2/xml/'
def __init__(self, user, token, fetcher):
self.user = user
self.token = token
self.fetcher = fetcher
def _raw_fetch(self, path):
p = self.BASE_URL + path
args = ''
if self.user and self.token:
params = '&'.join(['login=' + urllib.quote(self.user),
'token=' + urllib.quote(self.token)])
if '?' in path:
p += params
else:
p += '?' + params
return self.fetcher(p).read()
def _fetch(self, path):
return xml.dom.minidom.parseString(self._raw_fetch(path))
def _post(self, path, **kwargs):
p = {'login': self.user, 'token': self.token}
p.update(kwargs)
return self.fetcher(self.BASE_URL + path, urllib.urlencode(p)).read()
def _parsed(self, path):
doc = self._fetch(path)
return _parse(doc.documentElement)
def _posted(self,path,**kwargs):
stuff = self._post(path,**kwargs)
doc = xml.dom.minidom.parseString(stuff)
return _parse(doc.documentElement)
class UserEndpoint(BaseEndpoint):
def search(self, query):
"""Search for a user."""
return self._parsed('user/search/' + query)
def show(self, username):
"""Get the info for a user."""
return self._parsed('user/show/' + username)
def keys(self):
"""Get the public keys for a user."""
return self._parsed('user/keys')
def removeKey(self, keyId):
"""Remove the key with the given ID (as retrieved from keys)"""
self._post('user/key/remove', id=keyId)
def addKey(self, name, key):
"""Add an ssh key."""
self._post('user/key/add', name=name, key=key)
class RepositoryEndpoint(BaseEndpoint):
def forUser(self, username):
"""Get the repositories for the given user."""
return self._parsed('repos/show/' + username)
def branches(self, user, repo):
"""List the branches for a repo."""
doc = self._fetch("repos/show/" + user + "/" + repo + "/branches")
rv = {}
for c in doc.documentElement.childNodes:
if c.nodeType != xml.dom.Node.TEXT_NODE:
rv[c.localName] = str(c.firstChild.data)
return rv
def search(self, term):
"""Search for repositories."""
return self._parsed('repos/search/' + urllib.quote_plus(term))
def show(self, user, repo):
"""Lookup an individual repository."""
return self._parsed('/'.join(['repos', 'show', user, repo]))
def watch(self, user, repo):
"""Watch a repository."""
self._post('repos/watch/' + user + '/' + repo)
def unwatch(self, user, repo):
"""Stop watching a repository."""
self._post('repos/unwatch/' + user + '/' + repo)
def watched(self, user):
"""Get watched repositories of a user."""
return self._parsed('repos/watched/' + user)
def network(self, user, repo):
"""Get the network for a given repo."""
return self._parsed('repos/show/' + user + '/' + repo + '/network')
def setVisible(self, repo, public=True):
"""Set the visibility of the given repository (owned by the current user)."""
if public:
path = 'repos/set/public/' + repo
else:
path = 'repos/set/private/' + repo
self._post(path)
def create(self, name, description='', homepage='', public=1):
"""Create a new repository."""
self._post('repos/create', name=name, description=description,
homepage=homepage, public=str(public))
def delete(self, repo):
"""Delete a repository."""
self._post('repos/delete/' + repo)
def fork(self, user, repo):
"""Fork a user's repo."""
self._post('repos/fork/' + user + '/' + repo)
def collaborators(self, user, repo):
"""Find all of the collaborators of one of your repositories."""
return self._parsed('repos/show/%s/%s/collaborators' % (user, repo))
def addCollaborator(self, repo, username):
"""Add a collaborator to one of your repositories."""
self._post('repos/collaborators/' + repo + '/add/' + username)
def removeCollaborator(self, repo, username):
"""Remove a collaborator from one of your repositories."""
self._post('repos/collaborators/' + repo + '/remove/' + username)
def collaborators_all(self):
"""Find all of the collaborators of every of your repositories.
Returns a dictionary with reponame as key and a list of collaborators as value."""
ret = {}
for reponame in (rp.name for rp in self.forUser(self.user)):
ret[reponame] = self.collaborators(self.user, reponame)
return ret
def addCollaborator_all(self, username):
"""Add a collaborator to all of your repositories."""
for reponame in (rp.name for rp in self.forUser(self.user)):
self.addCollaborator(reponame, username)
def removeCollaborator_all(self, username):
"""Remove a collaborator from all of your repositories."""
for reponame in (rp.name for rp in self.forUser(self.user)):
self.removeCollaborator(reponame, username)
def deployKeys(self, repo):
"""List the deploy keys for the given repository.
The repository must be owned by the current user."""
return self._parsed('repos/keys/' + repo)
def addDeployKey(self, repo, title, key):
"""Add a deploy key to a repository."""
self._post('repos/key/' + repo + '/add', title=title, key=key)
def removeDeployKey(self, repo, keyId):
"""Remove a deploy key."""
self._post('repos/key/' + repo + '/remove', id=keyId)
class CommitEndpoint(BaseEndpoint):
def forBranch(self, user, repo, branch='master'):
"""Get the commits for the given branch."""
return self._parsed('/'.join(['commits', 'list', user, repo, branch]))
def forFile(self, user, repo, path, branch='master'):
"""Get the commits for the given file within the given branch."""
return self._parsed('/'.join(['commits', 'list', user, repo, branch, path]))
@with_temporary_mappings({'removed': _parseArray,
'added': _parseArray,
'modified': Modification,
'diff': _string_parser,
'filename': _string_parser})
def show(self, user, repo, sha):
"""Get an individual commit."""
c = self._parsed('/'.join(['commits', 'show', user, repo, sha]))
# Some fixup due to weird XML structure
if hasattr(c, 'removed'):
c.removed = [i[0] for i in c.removed]
if hasattr(c, 'added'):
c.added = [i[0] for i in c.added]
return c
class IssuesEndpoint(BaseEndpoint):
@with_temporary_mappings({'user': None})
def list(self, user, repo, state='open'):
"""Get the list of issues for the given repo in the given state."""
return self._parsed('/'.join(['issues', 'list', user, repo, state]))
@with_temporary_mappings({'user': None})
def show(self, user, repo, issue_id):
"""Show an individual issue."""
return self._parsed('/'.join(['issues', 'show', user, repo, str(issue_id)]))
def add_label(self, user, repo, issue_id, label):
"""Add a label to an issue."""
self._post('issues/label/add/' + user + '/'
+ repo + '/' + label + '/' + str(issue_id))
def remove_label(self, user, repo, issue_id, label):
"""Remove a label from an issue."""
self._post('issues/label/remove/' + user + '/'
+ repo + '/' + label + '/' + str(issue_id))
def close(self, user, repo, issue_id):
"""Close an issue."""
self._post('/'.join(['issues', 'close', user, repo, str(issue_id)]))
def reopen(self, user, repo, issue_id):
"""Reopen an issue."""
self._post('/'.join(['issues', 'reopen', user, repo, str(issue_id)]))
def new(self, user, repo, title, body=''):
"""Create a new issue."""
return self._posted('/'.join(['issues', 'open', user, repo]),
title=title, body=body)
def edit(self, user, repo, issue_id, title, body):
"""Create a new issue."""
self._post('/'.join(['issues', 'edit', user, repo, str(issue_id)]),
title=title, body=body)
class ObjectsEndpoint(BaseEndpoint):
@with_temporary_mappings({'tree': Tree, 'type': _string_parser})
def tree(self, user, repo, t):
"""Get the given tree from the given repo."""
tl = self._parsed('/'.join(['tree', 'show', user, repo, t]))
return dict([(t.name, t) for t in tl])
@with_temporary_mappings({'blob': Blob})
def blob(self, user, repo, t, fn):
return self._parsed('/'.join(['blob', 'show', user, repo, t, fn]))
def raw_blob(self, user, repo, sha):
"""Get a raw blob from a repo."""
path = 'blob/show/%s/%s/%s' % (user, repo, sha)
return self._raw_fetch(path)
class GitHub(object):
"""Interface to github."""
def __init__(self, user=None, token=None, fetcher=default_fetcher):
self.user = user
self.token = token
self.fetcher = fetcher
@property
def users(self):
"""Get access to the user API."""
return UserEndpoint(self.user, self.token, self.fetcher)
@property
def repos(self):
"""Get access to the user API."""
return RepositoryEndpoint(self.user, self.token, self.fetcher)
@property
def commits(self):
return CommitEndpoint(self.user, self.token, self.fetcher)
@property
def issues(self):
return IssuesEndpoint(self.user, self.token, self.fetcher)
@property
def objects(self):
return ObjectsEndpoint(self.user, self.token, self.fetcher)

88
lib/pygithub/githubsync.py
View File

@@ -1,88 +0,0 @@
#!/usr/bin/env python
#
# Copyright (c) 2005-2008 Dustin Sallings <dustin@spy.net>
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.
#
# <http://www.opensource.org/licenses/mit-license.php>
"""
Grab all of a user's projects from github.
"""
import os
import sys
import subprocess
import github
def check_for_old_format(path, url):
p = subprocess.Popen(['git', '--git-dir=' + path, 'config',
'remote.origin.fetch'], stdout = subprocess.PIPE)
stdout, stderr = p.communicate()
if stdout.strip() != '+refs/*:refs/*':
print "Not properly configured for mirroring, repairing."
subprocess.call(['git', '--git-dir=' + path, 'remote', 'rm', 'origin'])
add_mirror(path, url)
def add_mirror(path, url):
subprocess.call(['git', '--git-dir=' + path, 'remote', 'add', '--mirror',
'origin', url])
def sync(path, url, repo_name):
p = os.path.join(path, repo_name) + ".git"
print "Syncing %s -> %s" % (repo_name, p)
if not os.path.exists(p):
subprocess.call(['git', 'clone', '--bare', url, p])
add_mirror(p, url)
check_for_old_format(p, url)
subprocess.call(['git', '--git-dir=' + p, 'fetch', '-f'])
def sync_user_repo(path, repo):
sync(path, "git://github.com/%s/%s" % (repo.owner, repo.name), repo.name)
def usage():
sys.stderr.write("Usage: %s username destination_url\n" % sys.argv[0])
sys.stderr.write(
"""Ensures you've got the latest stuff for the given user.
Also, if the file $HOME/.github-private exists, it will be read for
additional projects.
Each line must be a simple project name (e.g. py-github), a tab character,
and a git URL.
""")
if __name__ == '__main__':
try:
user, path = sys.argv[1:]
except ValueError:
usage()
exit(1)
privfile = os.path.join(os.getenv("HOME"), ".github-private")
if os.path.exists(privfile):
f = open(privfile)
for line in f:
name, url = line.strip().split("\t")
sync(path, url, name)
gh = github.GitHub()
for repo in gh.repos.forUser(user):
sync_user_repo(path, repo)

493
lib/pygithub/githubtest.py
View File

@@ -1,493 +0,0 @@
#!/usr/bin/env python
#
# Copyright (c) 2005-2008 Dustin Sallings <dustin@spy.net>
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.
#
# <http://www.opensource.org/licenses/mit-license.php>
"""
Defines and runs unittests.
"""
import urllib
import hashlib
import unittest
import github
class BaseCase(unittest.TestCase):
def _gh(self, expUrl, filename):
def opener(url):
self.assertEquals(expUrl, url)
return open(filename)
return github.GitHub(fetcher=opener)
def _agh(self, expUrl, u, t, filename):
def opener(url):
self.assertEquals(expUrl, url + '?login=' + u + '&token=' + t)
return open(filename)
return github.GitHub(fetcher=opener)
def _ghp(self, expUrl, u, t, **kv):
def opener(url, data):
h = {'login': u, 'token': t}
h.update(kv)
self.assertEquals(github.BaseEndpoint.BASE_URL + expUrl, url)
self.assertEquals(sorted(data.split('&')),
sorted(urllib.urlencode(h).split('&')))
return github.GitHub(u, t, fetcher=opener)
class UserTest(BaseCase):
def __loadUserSearch(self):
return self._gh('http://github.com/api/v2/xml/user/search/dustin',
'data/user.search.xml').users.search('dustin')
def __loadUser(self, which, u=None, p=None):
if u:
return self._agh('http://github.com/api/v2/xml/user/show/dustin'
+ '?login=' + u + '&token=' + p,
u, p, 'data/' + which).users.show('dustin')
else:
return self._gh('http://github.com/api/v2/xml/user/show/dustin',
'data/' + which).users.show('dustin')
def testUserSearch(self):
"""Test the base properties of the user object."""
u = self.__loadUserSearch()[0]
self.assertEquals("Dustin Sallings", u.fullname)
self.assertEquals("dustin", u.name)
self.assertEquals("dustin@spy.net", u.email)
self.assertEquals("Santa Clara, CA", u.location)
self.assertEquals("Ruby", u.language)
self.assertEquals(35, u.actions)
self.assertEquals(77, u.repos)
self.assertEquals(78, u.followers)
self.assertEquals('user-1779', u.id)
self.assertAlmostEquals(12.231684, u.score)
self.assertEquals('user', u.type)
self.assertEquals('2008-02-29T17:59:09Z', u.created)
self.assertEquals('2009-03-19T09:15:24.663Z', u.pushed)
self.assertEquals("<<User dustin>>", repr(u))
def testUserPublic(self):
"""Test the user show API with no authentication."""
u = self.__loadUser('user.public.xml')
self.assertEquals("Dustin Sallings", u.name)
# self.assertEquals(None, u.company)
self.assertEquals(10, u.following_count)
self.assertEquals(21, u.public_gist_count)
self.assertEquals(81, u.public_repo_count)
self.assertEquals('http://bleu.west.spy.net/~dustin/', u.blog)
self.assertEquals(1779, u.id)
self.assertEquals(82, u.followers_count)
self.assertEquals('dustin', u.login)
self.assertEquals('Santa Clara, CA', u.location)
self.assertEquals('dustin@spy.net', u.email)
self.assertEquals('2008-02-29T09:59:09-08:00', u.created_at)
def testUserPrivate(self):
"""Test the user show API with extra info from auth."""
u = self.__loadUser('user.private.xml', 'dustin', 'blahblah')
self.assertEquals("Dustin Sallings", u.name)
# self.assertEquals(None, u.company)
self.assertEquals(10, u.following_count)
self.assertEquals(21, u.public_gist_count)
self.assertEquals(81, u.public_repo_count)
self.assertEquals('http://bleu.west.spy.net/~dustin/', u.blog)
self.assertEquals(1779, u.id)
self.assertEquals(82, u.followers_count)
self.assertEquals('dustin', u.login)
self.assertEquals('Santa Clara, CA', u.location)
self.assertEquals('dustin@spy.net', u.email)
self.assertEquals('2008-02-29T09:59:09-08:00', u.created_at)
# Begin private data
self.assertEquals("micro", u.plan.name)
self.assertEquals(1, u.plan.collaborators)
self.assertEquals(614400, u.plan.space)
self.assertEquals(5, u.plan.private_repos)
self.assertEquals(155191, u.disk_usage)
self.assertEquals(6, u.collaborators)
self.assertEquals(4, u.owned_private_repo_count)
self.assertEquals(5, u.total_private_repo_count)
self.assertEquals(0, u.private_gist_count)
def testKeysList(self):
"""Test key listing."""
kl = self._agh('http://github.com/api/v2/xml/user/keys?login=dustin&token=blahblah',
'dustin', 'blahblah', 'data/keys.xml').users.keys()
self.assertEquals(7, len(kl))
k = kl[0]
self.assertEquals('some key', k.title)
self.assertEquals(2181, k.id)
self.assertEquals(549, k.key.find('cdEXwCSjAIFp8iRqh3GOkxGyFSc25qv/MuOBg=='))
def testRemoveKey(self):
"""Remove a key."""
self._ghp('user/key/remove',
'dustin', 'p', id=828).users.removeKey(828)
def testAddKey(self):
"""Add a key."""
self._ghp('user/key/add',
'dustin', 'p', name='my key', key='some key').users.addKey(
'my key', 'some key')
class RepoTest(BaseCase):
def __loadUserRepos(self):
return self._gh('http://github.com/api/v2/xml/repos/show/verbal',
'data/repos.xml').repos.forUser('verbal')
def testUserRepoList(self):
"""Get a list of repos for a user."""
rs = self.__loadUserRepos()
self.assertEquals(10, len(rs))
r = rs[0]
self.assertEquals('A beanstalk client for the twisted network framework.',
r.description)
self.assertEquals(2, r.watchers)
self.assertEquals(0, r.forks)
self.assertEquals('beanstalk-client-twisted', r.name)
self.assertEquals(False, r.private)
self.assertEquals('http://github.com/verbal/beanstalk-client-twisted',
r.url)
self.assertEquals(True, r.fork)
self.assertEquals('verbal', r.owner)
# XXX: Can't parse empty elements. :(
# self.assertEquals('', r.homepage)
def testRepoSearch(self):
"""Test searching a repository."""
rl = self._gh('http://github.com/api/v2/xml/repos/search/ruby+testing',
'data/repos.search.xml').repos.search('ruby testing')
self.assertEquals(12, len(rl))
r = rl[0]
self.assertEquals('synthesis', r.name)
self.assertAlmostEquals(0.3234576, r.score, 4)
self.assertEquals(4656, r.actions)
self.assertEquals(2048, r.size)
self.assertEquals('Ruby', r.language)
self.assertEquals(26, r.followers)
self.assertEquals('gmalamid', r.username)
self.assertEquals('repo', r.type)
self.assertEquals('repo-3555', r.id)
self.assertEquals(1, r.forks)
self.assertFalse(r.fork)
self.assertEquals('Ruby test code analysis tool employing a '
'"Synthesized Testing" strategy, aimed to reduce '
'the volume of slower, coupled, complex wired tests.',
r.description)
self.assertEquals('2009-01-08T13:45:06Z', r.pushed)
self.assertEquals('2008-03-11T23:38:04Z', r.created)
def testBranchList(self):
"""Test branch listing for a repo."""
bl = self._gh('http://github.com/api/v2/xml/repos/show/schacon/ruby-git/branches',
'data/repos.branches.xml').repos.branches('schacon', 'ruby-git')
self.assertEquals(4, len(bl))
self.assertEquals('ee90922f3da3f67ef19853a0759c1d09860fe3b3', bl['master'])
def testGetOneRepo(self):
"""Fetch an individual repository."""
r = self._gh('http://github.com/api/v2/xml/repos/show/schacon/grit',
'data/repo.xml').repos.show('schacon', 'grit')
self.assertEquals('Grit is a Ruby library for extracting information from a '
'git repository in an object oriented manner - this fork '
'tries to intergrate as much pure-ruby functionality as possible',
r.description)
self.assertEquals(68, r.watchers)
self.assertEquals(4, r.forks)
self.assertEquals('grit', r.name)
self.assertFalse(r.private)
self.assertEquals('http://github.com/schacon/grit', r.url)
self.assertTrue(r.fork)
self.assertEquals('schacon', r.owner)
self.assertEquals('http://grit.rubyforge.org/', r.homepage)
def testGetRepoNetwork(self):
"""Test network fetching."""
nl = self._gh('http://github.com/api/v2/xml/repos/show/dustin/py-github/network',
'data/network.xml').repos.network('dustin', 'py-github')
self.assertEquals(5, len(nl))
n = nl[0]
self.assertEquals('Python interface for talking to the github API',
n.description)
self.assertEquals('py-github', n.name)
self.assertFalse(n.private)
self.assertEquals('http://github.com/dustin/py-github', n.url)
self.assertEquals(30, n.watchers)
self.assertEquals(4, n.forks)
self.assertFalse(n.fork)
self.assertEquals('dustin', n.owner)
self.assertEquals('http://dustin.github.com/2008/12/29/github-sync.html',
n.homepage)
def testSetPublic(self):
"""Test setting a repo visible."""
self._ghp('repos/set/public/py-github', 'dustin', 'p').repos.setVisible(
'py-github')
def testSetPrivate(self):
"""Test setting a repo to private."""
self._ghp('repos/set/private/py-github', 'dustin', 'p').repos.setVisible(
'py-github', False)
def testCreateRepository(self):
"""Test creating a repository."""
self._ghp('repos/create', 'dustin', 'p',
name='testrepo',
description='woo',
homepage='',
public='1').repos.create(
'testrepo', description='woo')
def testDeleteRepo(self):
"""Test setting a repo to private."""
self._ghp('repos/delete/mytest', 'dustin', 'p').repos.delete('mytest')
def testFork(self):
"""Test forking'"""
self._ghp('repos/fork/someuser/somerepo', 'dustin', 'p').repos.fork(
'someuser', 'somerepo')
def testAddCollaborator(self):
"""Adding a collaborator."""
self._ghp('repos/collaborators/memcached/add/trondn',
'dustin', 'p').repos.addCollaborator('memcached', 'trondn')
def testRemoveCollaborator(self):
"""Removing a collaborator."""
self._ghp('repos/collaborators/memcached/remove/trondn',
'dustin', 'p').repos.removeCollaborator('memcached', 'trondn')
def testAddDeployKey(self):
"""Add a deploy key."""
self._ghp('repos/key/blah/add', 'dustin', 'p',
title='title', key='key').repos.addDeployKey('blah', 'title', 'key')
def testRemoveDeployKey(self):
"""Remove a deploy key."""
self._ghp('repos/key/blah/remove', 'dustin', 'p',
id=5).repos.removeDeployKey('blah', 5)
class CommitTest(BaseCase):
def testCommitList(self):
"""Test commit list."""
cl = self._gh('http://github.com/api/v2/xml/commits/list/mojombo/grit/master',
'data/commits.xml').commits.forBranch('mojombo', 'grit')
self.assertEquals(30, len(cl))
c = cl[0]
self.assertEquals("Regenerated gemspec for version 1.1.1", c.message)
self.assertEquals('4ac4acab7fd9c7fd4c0e0f4ff5794b0347baecde', c.id)
self.assertEquals('94490563ebaf733cbb3de4ad659eb58178c2e574', c.tree)
self.assertEquals('2009-03-31T09:54:51-07:00', c.committed_date)
self.assertEquals('2009-03-31T09:54:51-07:00', c.authored_date)
self.assertEquals('http://github.com/mojombo/grit/commit/4ac4acab7fd9c7fd4c0e0f4ff5794b0347baecde',
c.url)
self.assertEquals(1, len(c.parents))
self.assertEquals('5071bf9fbfb81778c456d62e111440fdc776f76c', c.parents[0].id)
self.assertEquals('Tom Preston-Werner', c.author.name)
self.assertEquals('tom@mojombo.com', c.author.email)
self.assertEquals('Tom Preston-Werner', c.committer.name)
self.assertEquals('tom@mojombo.com', c.committer.email)
def testCommitListForFile(self):
"""Test commit list for a file."""
cl = self._gh('http://github.com/api/v2/xml/commits/list/mojombo/grit/master/grit.gemspec',
'data/commits.xml').commits.forFile('mojombo', 'grit', 'grit.gemspec')
self.assertEquals(30, len(cl))
c = cl[0]
self.assertEquals("Regenerated gemspec for version 1.1.1", c.message)
self.assertEquals('4ac4acab7fd9c7fd4c0e0f4ff5794b0347baecde', c.id)
self.assertEquals('94490563ebaf733cbb3de4ad659eb58178c2e574', c.tree)
self.assertEquals('2009-03-31T09:54:51-07:00', c.committed_date)
self.assertEquals('2009-03-31T09:54:51-07:00', c.authored_date)
self.assertEquals('http://github.com/mojombo/grit/commit/4ac4acab7fd9c7fd4c0e0f4ff5794b0347baecde',
c.url)
self.assertEquals(1, len(c.parents))
self.assertEquals('5071bf9fbfb81778c456d62e111440fdc776f76c', c.parents[0].id)
self.assertEquals('Tom Preston-Werner', c.author.name)
self.assertEquals('tom@mojombo.com', c.author.email)
self.assertEquals('Tom Preston-Werner', c.committer.name)
self.assertEquals('tom@mojombo.com', c.committer.email)
def testIndividualCommit(self):
"""Grab a single commit."""
h = '4c86fa592fcc7cb685c6e9d8b6aebe8dcbac6b3e'
c = self._gh('http://github.com/api/v2/xml/commits/show/dustin/memcached/' + h,
'data/commit.xml').commits.show('dustin', 'memcached', h)
self.assertEquals(['internal_tests.c'], c.removed)
self.assertEquals(set(['cache.c', 'cache.h', 'testapp.c']), set(c.added))
self.assertEquals('Create a generic cache for objects of same size\n\n'
'The suffix pool could be thread-local and use the generic cache',
c.message)
self.assertEquals(6, len(c.modified))
self.assertEquals('.gitignore', c.modified[0].filename)
self.assertEquals(140, len(c.modified[0].diff))
self.assertEquals(['ee0c3d5ae74d0862b4d9990e2ad13bc79f8c34df'],
[p.id for p in c.parents])
self.assertEquals('http://github.com/dustin/memcached/commit/' + h, c.url)
self.assertEquals('Trond Norbye', c.author.name)
self.assertEquals('Trond.Norbye@sun.com', c.author.email)
self.assertEquals(h, c.id)
self.assertEquals('2009-04-17T16:15:52-07:00', c.committed_date)
self.assertEquals('2009-03-27T10:30:16-07:00', c.authored_date)
self.assertEquals('94b644163f6381a9930e2d7c583fae023895b903', c.tree)
self.assertEquals('Dustin Sallings', c.committer.name)
self.assertEquals('dustin@spy.net', c.committer.email)
def testWatchRepo(self):
"""Test watching a repo."""
self._ghp('repos/watch/dustin/py-github', 'dustin', 'p').repos.watch(
'dustin', 'py-github')
def testWatchRepo(self):
"""Test watching a repo."""
self._ghp('repos/unwatch/dustin/py-github', 'dustin', 'p').repos.unwatch(
'dustin', 'py-github')
class IssueTest(BaseCase):
def testListIssues(self):
"""Test listing issues."""
il = self._gh('http://github.com/api/v2/xml/issues/list/schacon/simplegit/open',
'data/issues.list.xml').issues.list('schacon', 'simplegit')
self.assertEquals(1, len(il))
i = il[0]
self.assertEquals('schacon', i.user)
self.assertEquals('2009-04-17T16:19:02-07:00', i.updated_at)
self.assertEquals('something', i.body)
self.assertEquals('new', i.title)
self.assertEquals(2, i.number)
self.assertEquals(0, i.votes)
self.assertEquals(1.0, i.position)
self.assertEquals('2009-04-17T16:18:50-07:00', i.created_at)
self.assertEquals('open', i.state)
def testShowIssue(self):
"""Show an individual issue."""
i = self._gh('http://github.com/api/v2/xml/issues/show/dustin/py-github/1',
'data/issues.show.xml').issues.show('dustin', 'py-github', 1)
self.assertEquals('dustin', i.user)
self.assertEquals('2009-04-17T18:37:04-07:00', i.updated_at)
self.assertEquals('http://develop.github.com/p/general.html', i.body)
self.assertEquals('Add auth tokens', i.title)
self.assertEquals(1, i.number)
self.assertEquals(0, i.votes)
self.assertEquals(1.0, i.position)
self.assertEquals('2009-04-17T17:00:58-07:00', i.created_at)
self.assertEquals('closed', i.state)
def testAddLabel(self):
"""Adding a label to an issue."""
self._ghp('issues/label/add/dustin/py-github/todo/33', 'd', 'pw').issues.add_label(
'dustin', 'py-github', 33, 'todo')
def testRemoveLabel(self):
"""Removing a label from an issue."""
self._ghp('issues/label/remove/dustin/py-github/todo/33',
'd', 'pw').issues.remove_label(
'dustin', 'py-github', 33, 'todo')
def testCloseIssue(self):
"""Closing an issue."""
self._ghp('issues/close/dustin/py-github/1', 'd', 'pw').issues.close(
'dustin', 'py-github', 1)
def testReopenIssue(self):
"""Reopening an issue."""
self._ghp('issues/reopen/dustin/py-github/1', 'd', 'pw').issues.reopen(
'dustin', 'py-github', 1)
def testCreateIssue(self):
"""Creating an issue."""
self._ghp('issues/open/dustin/py-github', 'd', 'pw',
title='test title', body='').issues.new(
'dustin', 'py-github', title='test title')
def testEditIssue(self):
"""Editing an existing issue."""
self._ghp('issues/edit/dustin/py-github/1', 'd', 'pw',
title='new title', body='new body').issues.edit(
'dustin', 'py-github', 1, 'new title', 'new body')
class ObjectTest(BaseCase):
def testTree(self):
"""Test tree fetching."""
h = '1ddd3f99f0b96019042239375b3ad4d45796ffba'
tl = self._gh('http://github.com/api/v2/xml/tree/show/dustin/py-github/' + h,
'data/tree.xml').objects.tree('dustin', 'py-github', h)
self.assertEquals(8, len(tl))
self.assertEquals('setup.py', tl['setup.py'].name)
self.assertEquals('6e290379ec58fa00ac9d1c2a78f0819a21397445',
tl['setup.py'].sha)
self.assertEquals('100755', tl['setup.py'].mode)
self.assertEquals('blob', tl['setup.py'].type)
self.assertEquals('src', tl['src'].name)
self.assertEquals('5fb9175803334c82b3fd66f1b69502691b91cf4f',
tl['src'].sha)
self.assertEquals('040000', tl['src'].mode)
self.assertEquals('tree', tl['src'].type)
def testBlob(self):
"""Test blob fetching."""
h = '1ddd3f99f0b96019042239375b3ad4d45796ffba'
blob = self._gh('http://github.com/api/v2/xml/blob/show/dustin/py-github/'
+ h + '/setup.py',
'data/blob.xml').objects.blob('dustin', 'py-github', h, 'setup.py')
self.assertEquals('setup.py', blob.name)
self.assertEquals(1842, blob.size)
self.assertEquals('6e290379ec58fa00ac9d1c2a78f0819a21397445', blob.sha)
self.assertEquals('100755', blob.mode)
self.assertEquals('text/plain', blob.mime_type)
self.assertEquals(1842, len(blob.data))
self.assertEquals(1641, blob.data.index('Production/Stable'))
def testRawBlob(self):
"""Test raw blob fetching."""
h = '6e290379ec58fa00ac9d1c2a78f0819a21397445'
blob = self._gh('http://github.com/api/v2/xml/blob/show/dustin/py-github/' + h,
'data/setup.py').objects.raw_blob('dustin', 'py-github', h)
self.assertEquals('e2dc8aea9ae8961f4f5923f9febfdd0a',
hashlib.md5(blob).hexdigest())
if __name__ == '__main__':
unittest.main()