Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
async def test_proxy(get_version, monkeypatch):
from nvchecker.source import session
async def fake_request(*args, proxy, **kwargs):
class fake_response():
status = 200
async def read():
return proxy.encode("ascii")
def release():
pass
return fake_response
monkeypatch.setattr(session, "nv_config", {"proxy": "255.255.255.255:65535"}, raising=False)
monkeypatch.setattr(aiohttp.ClientSession, "_request", fake_request)
assert await get_version("example", {"regex": "(.+)", "url": "deadbeef"}) == "255.255.255.255:65535"
assert await get_version("example", {"regex": "(.+)", "url": "deadbeef", "proxy": "0.0.0.0:0"}) == "0.0.0.0:0"
async def get_version(name, conf, **kwargs):
pkg = conf.get('debianpkg') or name
strip_release = conf.getboolean('strip-release', False)
suite = conf.get('suite') or "sid"
url = URL % {"pkgname": pkg, "suite": suite}
async with session.get(url) as res:
data = await res.json()
if not data.get('versions'):
logger.error('Debian package not found', name=name)
return
r = data['versions'][0]
if strip_release:
version = r['version'].split("-")[0]
else:
version = r['version']
return version
async def get_version(name, conf, **kwargs):
aurname = conf.get('aur') or name
use_last_modified = conf.getboolean('use_last_modified', False)
strip_release = conf.getboolean('strip-release', False)
async with session.get(AUR_URL, params={"v": 5, "type": "info", "arg[]": aurname}) as res:
data = await res.json()
if not data['results']:
logger.error('AUR upstream not found', name=name)
return
version = data['results'][0]['Version']
if use_last_modified:
version += '-' + datetime.utcfromtimestamp(data['results'][0]['LastModified']).strftime('%Y%m%d%H%M%S')
if strip_release and '-' in version:
version = version.rsplit('-', 1)[0]
return version
token = conf.get('token')
# Load token from environ
if token is None:
env_name = "NVCHECKER_GITLAB_TOKEN_" + host.upper().replace(".", "_").replace("/", "_")
token = os.environ.get(env_name)
# Load token from keyman
if token is None and 'keyman' in kwargs:
key_name = 'gitlab_' + host.lower().replace('.', '_').replace("/", "_")
token = kwargs['keyman'].get_key(key_name)
# Set private token if token exists.
headers = {}
if token:
headers["PRIVATE-TOKEN"] = token
async with session.get(url, headers=headers) as res:
data = await res.json()
if use_max_tag:
version = [tag["name"] for tag in data if tag["name"] not in ignored_tags]
else:
version = data[0]['created_at'].split('T', 1)[0].replace('-', '')
return version
async def get_latest_tag(name, conf, token):
repo = conf.get('github')
query = conf.get('query', '')
owner, reponame = repo.split('/')
headers = {
'Authorization': 'bearer %s' % token,
'Content-Type': 'application/json',
}
q = QUERY_LATEST_TAG.format(
owner = owner,
name = reponame,
query = query,
)
async with session.post(
GITHUB_GRAPHQL_URL,
headers = headers,
json = {'query': q},
) as res:
j = await res.json()
refs = j['data']['repository']['refs']['edges']
if not refs:
logger.error('no tag found', name=name)
return
return refs[0]['node']['name']
async def get_version(name, conf):
repo = conf.get('gcode_svn') or name
url = GCODE_URL % repo
async with session.get(url) as res:
data = await res.text()
m = GCODE_SVN_RE.search(data)
if m:
version = m.group(1)
else:
logger.error('%s: version not found.', name)
version = None
return name, version
async def get_version(name, conf, **kwargs):
pkg = conf.get('anitya')
url = URL.format(pkg = pkg)
async with session.get(url) as res:
data = await res.json()
return data['version']
async def get_version(name, conf, **kwargs):
pkg = conf.get('ubuntupkg') or name
strip_release = conf.getboolean('strip-release', False)
suite = conf.get('suite')
url = URL % pkg
if suite:
suite = "https://api.launchpad.net/1.0/ubuntu/" + suite
releases = []
while not releases:
async with session.get(url) as res:
data = await res.json()
if not data.get('entries'):
logger.error('Ubuntu package not found', name=name)
return
releases = [r for r in data["entries"] if r["status"] == "Published"]
if suite:
releases = [r for r in releases if r["distro_series_link"] == suite]
if "next_collection_link" not in data:
break
url = data["next_collection_link"]
if path:
parameters['path'] = path
url += '?' + urlencode(parameters)
headers = {
'Accept': 'application/vnd.github.quicksilver-preview+json',
}
if token:
headers['Authorization'] = 'token %s' % token
kwargs = {}
if conf.get('proxy'):
kwargs["proxy"] = conf.get("proxy")
if use_max_tag:
return await max_tag(partial(
session.get, headers=headers, **kwargs),
url, name, ignored_tags, include_tags_pattern,
max_page = conf.getint("max_page", 1),
)
async with session.get(url, headers=headers, **kwargs) as res:
logger.debug('X-RateLimit-Remaining',
n=res.headers.get('X-RateLimit-Remaining'))
data = await res.json()
if use_latest_release:
if 'tag_name' not in data:
logger.error('No tag found in upstream repository.',
name=name)
return
version = data['tag_name']
async def get_version(name, conf, **kwargs):
repo = conf.get(confkey) or name
url = urlpat % repo
kwargs = {}
if conf.get('proxy'):
kwargs["proxy"] = conf.get('proxy')
async with session.get(url, **kwargs) as res:
data = await res.json(content_type=None)
version = version_from_json(data)
return version