fixed flake8 issues

parent 2e752095
Pipeline #60356 failed with stages
in 1 minute and 44 seconds
......@@ -5,9 +5,7 @@
"""Utilities for the GWDataFind Server
"""
import logging
import re
import socket
import threading
import time
from collections import defaultdict
......
......@@ -7,8 +7,6 @@
import os.path
from configobj import ConfigObj # pylint: disable=unused-import
def get_config_path(basename='gwdatafind-server.ini'):
"""Try and locate the given basename file in a set of standard directories
......
# -*- coding: utf-8 -*-
# Copyright (2019) Duncan Macleod
# Licensed under GPLv3+ - see LICENSE
"""Utilities for the GWDataFind Server
"""
......@@ -74,7 +74,8 @@ def find_types(ext, site):
sites = ecache.keys()
else:
sites = [site]
return [tag for site in sites for tag in current_app.get_cache_data(ext, site)]
return [tag for site in sites for
tag in current_app.get_cache_data(ext, site)]
@blueprint.route('<ext>/<site>/<tag>/segments.json')
......@@ -85,12 +86,12 @@ def find_all_times(ext, site, tag):
span = segmentlist([segment(0., INF)])
return reduce(
operator.or_,
(segs & span for segs in current_app.get_cache_data(ext, site, tag).values()),
(segs & span for
segs in current_app.get_cache_data(ext, site, tag).values()),
segmentlist(),
)
@blueprint.route('<ext>/<site>/<tag>/segments/<int:start>,<int:end>.json')
@as_json
def find_times(ext, site, tag, start, end):
......@@ -99,7 +100,8 @@ def find_times(ext, site, tag, start, end):
span = segmentlist([segment(float(start), float(end))])
return reduce(
operator.or_,
(segs & span for segs in current_app.get_cache_data(ext, site, tag).values()),
(segs & span for segs in
current_app.get_cache_data(ext, site, tag).values()),
segmentlist(),
)
......@@ -153,7 +155,11 @@ def _find_urls(ext, site, tag, start, end, urltype=None, match=None,
lfns = defaultdict(list)
maxgps = -1e9 # something absurdly old
print("TEST")
for (path, cdur), seglist in current_app.get_cache_data(ext, site, tag).items():
for (path, cdur), seglist in current_app.get_cache_data(
ext,
site,
tag,
).items():
# if running a 'latest' URL search, restrict the search to
# the most recent available segment for this frametype
print(path, cdur, seglist, search)
......
[metadata]
license_files = LICENSE
[flake8]
exclude =
__pycache__,
.eggs/,
.git/,
build/,
docs/,
per-file-ignores =
__init__.py:F401
......@@ -2,8 +2,8 @@
# Copyright (2019) Duncan Macleod
# Licensed under GPLv3+ - see LICENSE
import os.path
import re
from pathlib import Path
from setuptools import (setup, find_packages)
......@@ -15,7 +15,7 @@ def find_version(path):
"""
with open(path, "r") as fp:
version_file = fp.read()
version_match = re.search(r"^__version__ = ["\"]([^"\"]*)["\"]",
version_match = re.search(r"^__version__ = ['\"]([^'\"]*)['\"]",
version_file, re.M)
if version_match:
return version_match.group(1)
......
......@@ -16,7 +16,8 @@ from gwdatafind_server import create_app
@pytest.fixture
def app():
with mock.patch.dict(os.environ, {"LDR_LOCATION": str(Path(__file__).parent)}):
with mock.patch.dict(os.environ,
{"LDR_LOCATION": str(Path(__file__).parent)}):
yield create_app()
......
......@@ -2,21 +2,16 @@
# Copyright (2019) Duncan Macleod
# Licensed under GPLv3+ - see LICENSE
import os
from pathlib import Path
from unittest import mock
import pytest
import gwdatafind_server
from gwdatafind_server import views
# -- test the app ---------------------
@mock.patch.dict(os.environ, {"LDR_LOCATION": str(Path(__file__).parent)})
def test_shutdown():
app = gwdatafind_server.create_app()
def test_shutdown(app):
assert app.manager.is_alive()
app.shutdown()
assert not app.manager.is_alive()
......@@ -54,13 +49,17 @@ def test_find_times(client):
"1000000007,1000000013.json",
)
assert resp.status_code == 200
assert resp.json == [[1000000007, 1000000008], [1000000012, 1000000013]]
assert resp.json == [
[1000000007, 1000000008],
[1000000012, 1000000013],
]
@mock.patch.object(views, "_DEFAULT_GSIFTP_HOST", new="testhost")
def test_find_url(client):
resp = client.get(
"/LDR/services/data/v1/h5/L/L1_TEST_1/L-L1_TEST_1-1000000004-4.h5.json",
"/LDR/services/data/v1"
"/h5/L/L1_TEST_1/L-L1_TEST_1-1000000004-4.h5.json",
)
assert resp.status_code == 200
assert resp.json == [
......@@ -85,7 +84,8 @@ def test_find_urls(client):
def test_find_urls_urltype(client):
resp = client.get(
"/LDR/services/data/v1/gwf/L/L1_TEST_1/1000000004,1000000016/file.json",
"/LDR/services/data/v1"
"/gwf/L/L1_TEST_1/1000000004,1000000016/file.json",
)
assert resp.status_code == 200
assert resp.json == [
......@@ -106,7 +106,7 @@ def test_find_latest(client):
]
def test_find_latest(client):
def test_find_latest_urltype(client):
resp = client.get(
"/LDR/services/data/v1/gwf/L/L1_TEST_1/latest/file.json",
)
......
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