Skip to content
Snippets Groups Projects

Compare revisions

Changes are shown as if the source revision was being merged into the target revision. Learn more about comparing revisions.

Source

Select target project
No results found

Target

Select target project
  • duncan.meacher/gwdatafind
  • duncanmmacleod/gwdatafind
  • computing/gwdatafind/client
3 results
Show changes
Commits on Source (9)
include:
# -- Python -------------------------
- component: $CI_SERVER_FQDN/computing/gitlab/components/python/all@1
- component: $CI_SERVER_FQDN/computing/gitlab/components/python/all@2
inputs:
code_quality_analyzer: ruff
install_extra: "test"
run_advanced_sast: true
- component: $CI_SERVER_FQDN/computing/gitlab/components/python/type-checking@2
# -- Debian packaging ---------------
......@@ -36,7 +37,7 @@ include:
# -- Red Hat packaging --------------
- component: $CI_SERVER_FQDN/computing/gitlab/components/redhat/all@1
- component: $CI_SERVER_FQDN/computing/gitlab/components/redhat/all@2
inputs:
needs: [sdist]
redhat_versions:
......
......@@ -14,14 +14,14 @@ project = "gwdatafind"
copyright = "2018-2025, Cardiff University"
author = "Duncan Macleod"
release = gwdatafind.__version__
version = re.split(r'[\w-]', gwdatafind.__version__)[0]
version = re.split(r"[\w-]", gwdatafind.__version__)[0]
# -- config
source_suffix = '.rst'
master_doc = 'index'
source_suffix = ".rst"
master_doc = "index"
default_role = 'obj'
default_role = "obj"
# -- theme
......
......@@ -85,6 +85,6 @@ from igwn_auth_utils.requests import Session
from .ui import *
__author__ = 'Duncan Macleod <duncan.macleod@ligo.org>'
__credits__ = 'Scott Koranda <scott.koranda@ligo.org>'
__version__ = '1.2.0'
__author__ = "Duncan Macleod <duncan.macleod@ligo.org>"
__credits__ = "Scott Koranda <scott.koranda@ligo.org>"
__version__ = "1.2.0"
......@@ -15,8 +15,7 @@
# You should have received a copy of the GNU General Public License
# along with GWDataFind. If not, see <http://www.gnu.org/licenses/>.
"""Query a GWDataFind server for information.
"""
"""Query a GWDataFind server for information."""
import argparse
import re
......@@ -34,8 +33,8 @@ from .io import (
)
from .utils import get_default_host
__author__ = 'Duncan Macleod <duncan.macleod@ligo.org>'
__credits__ = 'Scott Koranda, The LIGO Scientific Collaboration'
__author__ = "Duncan Macleod <duncan.macleod@ligo.org>"
__credits__ = "Scott Koranda, The LIGO Scientific Collaboration"
# -- command line parsing -----------------------------------------------------
......@@ -44,7 +43,7 @@ class DataFindFormatter(
argparse.ArgumentDefaultsHelpFormatter,
argparse.RawDescriptionHelpFormatter,
):
pass
"""Custom `argparse.Formatter` for GWDataFind."""
class DataFindArgumentParser(argparse.ArgumentParser):
......@@ -52,16 +51,15 @@ class DataFindArgumentParser(argparse.ArgumentParser):
Mainly to handle the legacy mutually-exclusive optional arguments...
"""
def __init__(self, *args, **kwargs):
"""Create a new `DataFindArgumentParser`.
"""
"""Create a new `DataFindArgumentParser`."""
kwargs.setdefault("formatter_class", DataFindFormatter)
super().__init__(*args, **kwargs)
self._optionals.title = "Optional arguments"
def parse_args(self, *args, **kwargs):
"""Parse arguments with an extra sanity check.
"""
"""Parse arguments with an extra sanity check."""
args = super().parse_args(*args, **kwargs)
args.show_urls = not any((args.ping, args.show_observatories,
args.show_types, args.show_times,
......@@ -96,12 +94,11 @@ class DataFindArgumentParser(argparse.ArgumentParser):
"--gps-end-time time all must be given when querying "
"for file URLs")
if namespace.gaps and not namespace.show_urls:
self.error('-g/--gaps only allowed when querying for file URLs')
self.error("-g/--gaps only allowed when querying for file URLs")
def command_line():
"""Build an `~argparse.ArgumentParser` for the `gwdatafind` CLI.
"""
"""Build an `~argparse.ArgumentParser` for the `gwdatafind` CLI."""
try:
defhost = get_default_host()
except ValueError:
......@@ -276,7 +273,7 @@ def command_line():
default="file",
help=(
"return only URLs with a particular scheme or head "
"such as \"file\" or \"gsiftp\""
'such as "file" or "gsiftp"'
),
)
oargs.add_argument(
......@@ -399,10 +396,10 @@ def show_times(args, out):
host=args.server,
ext=args.extension,
)
print('# seg\tstart \tstop \tduration', file=out)
print("# seg\tstart \tstop \tduration", file=out)
for i, seg in enumerate(seglist):
print(
f'{i}\t{seg[0]:10}\t{seg[1]:10}\t{abs(seg)}',
f"{i}\t{seg[0]:10}\t{seg[1]:10}\t{abs(seg)}",
file=out,
)
......@@ -427,7 +424,7 @@ def latest(args, out):
args.observatory,
args.type,
urltype=args.url_type,
on_missing='warn',
on_missing="warn",
host=args.server,
ext=args.extension,
)
......@@ -453,7 +450,7 @@ def filename(args, out):
cache = ui.find_url(
args.filename,
urltype=args.url_type,
on_missing='warn',
on_missing="warn",
host=args.server,
)
return postprocess_cache(cache, args, out)
......@@ -483,7 +480,7 @@ def show_urls(args, out):
match=args.match,
urltype=args.url_type,
host=args.server,
on_gaps='ignore',
on_gaps="ignore",
ext=args.extension,
)
return postprocess_cache(cache, args, out)
......@@ -496,10 +493,10 @@ def postprocess_cache(urls, args, out):
in the requested format, then prints gaps to stderr if requested.
"""
# if searching for SFTs replace '.gwf' file suffix with '.sft'
if re.search(r'_\d+SFT(\Z|_)', str(args.type)):
gwfreg = re.compile(r'\.gwf\Z')
if re.search(r"_\d+SFT(\Z|_)", str(args.type)):
gwfreg = re.compile(r"\.gwf\Z")
for i, url in enumerate(urls):
urls[i] = gwfreg.sub('.sft', url)
urls[i] = gwfreg.sub(".sft", url)
cache = lal_cache(urls)
......@@ -524,15 +521,14 @@ def postprocess_cache(urls, args, out):
# -- CLI ----------------------------------------------------------------------
def main(args=None):
"""Run the thing.
"""
"""Run the thing."""
# parse command line
parser = command_line()
opts = parser.parse_args(args=args)
# open output
if opts.output_file:
out = open(opts.output_file, 'w')
out = open(opts.output_file, "w")
else:
out = sys.stdout
......@@ -557,5 +553,5 @@ def main(args=None):
out.close()
if __name__ == '__main__': # pragma: no-cover
if __name__ == "__main__": # pragma: no-cover
sys.exit(main())
......@@ -23,7 +23,7 @@ to execute various GWDataFind queries.
from functools import wraps
from os.path import basename
__author__ = 'Duncan Macleod <duncan.macleod@ligo.org>'
__author__ = "Duncan Macleod <duncan.macleod@ligo.org>"
DEFAULT_EXT = "gwf"
DEFAULT_SERVICE_PREFIX = "LDR/services/data/v1"
......@@ -45,22 +45,19 @@ def _prefix(func):
@_prefix
def ping_path(ext=DEFAULT_EXT):
"""Return the API path to ping the server.
"""
"""Return the API path to ping the server."""
return f"{ext}/H/R/1,2"
@_prefix
def find_observatories_path(ext=DEFAULT_EXT):
"""Return the API path to query for all observatories.
"""
"""Return the API path to query for all observatories."""
return f"{ext}.json"
@_prefix
def find_types_path(site=None, ext=DEFAULT_EXT):
"""Return the API path to query for datasets for one or all sites.
"""
"""Return the API path to query for datasets for one or all sites."""
if site:
return f"{ext}/{site}.json"
return f"{ext}/all.json"
......@@ -68,8 +65,7 @@ def find_types_path(site=None, ext=DEFAULT_EXT):
@_prefix
def find_times_path(site, frametype, start, end, ext=DEFAULT_EXT):
"""Return the API path to query for data availability segments.
"""
"""Return the API path to query for data availability segments."""
if start is None and end is None:
return f"{ext}/{site}/{frametype}/segments.json"
return f"{ext}/{site}/{frametype}/segments/{start},{end}.json"
......@@ -77,8 +73,7 @@ def find_times_path(site, frametype, start, end, ext=DEFAULT_EXT):
@_prefix
def find_url_path(framefile):
"""Return the API path to query for the URL of a specific filename.
"""
"""Return the API path to query for the URL of a specific filename."""
filename = basename(framefile)
ext = filename.split(".", 1)[1]
site, frametype, _ = filename.split("-", 2)
......@@ -87,8 +82,7 @@ def find_url_path(framefile):
@_prefix
def find_latest_path(site, frametype, urltype, ext=DEFAULT_EXT):
"""Return the API path to query for the latest file in a dataset.
"""
"""Return the API path to query for the latest file in a dataset."""
stub = f"{ext}/{site}/{frametype}/latest"
if urltype:
return f"{stub}/{urltype}.json"
......@@ -105,8 +99,7 @@ def find_urls_path(
match=None,
ext=DEFAULT_EXT,
):
"""Return the API path to query for all URLs for a dataset in an interval.
"""
"""Return the API path to query for all URLs for a dataset in an interval."""
stub = f"{ext}/{site}/{frametype}/{start},{end}"
if urltype:
path = f"{stub}/{urltype}.json"
......
......@@ -20,7 +20,7 @@
import warnings
__author__ = 'Duncan Macleod <duncan.macleod@ligo.org>'
__author__ = "Duncan Macleod <duncan.macleod@ligo.org>"
# always present warnings during testing
warnings.simplefilter('always')
warnings.simplefilter("always")
......@@ -15,8 +15,7 @@
# You should have received a copy of the GNU General Public License
# along with GWDataFind. If not, see <http://www.gnu.org/licenses/>.
"""I/O (mainly O) routines for GWDataFind.
"""
"""I/O (mainly O) routines for GWDataFind."""
import os.path
from collections import namedtuple
......@@ -25,17 +24,19 @@ from urllib.parse import urlparse
from .utils import filename_metadata
__author__ = 'Duncan Macleod <duncan.macleod@ligo.org>'
__author__ = "Duncan Macleod <duncan.macleod@ligo.org>"
class LalCacheEntry(
namedtuple('CacheEntry', ('obs', 'tag', 'segment', 'url')),
namedtuple("CacheEntry", ("obs", "tag", "segment", "url")),
):
"""Simplified version of `lal.utils.CacheEntry`.
This is provided so that we don't have to depend on lalsuite.
"""
def __str__(self):
"""Return a `str` representation of this `LalCacheEntry`."""
seg = self.segment
return " ".join(map(str, (
self.obs,
......@@ -47,8 +48,7 @@ class LalCacheEntry(
@classmethod
def from_url(cls, url, **kwargs):
"""Create a new `LalCacheEntry` from a URL that follows LIGO-T050017.
"""
"""Create a new `LalCacheEntry` from a URL that follows LIGO-T050017."""
obs, tag, seg = filename_metadata(url)
return cls(obs, tag, seg, url)
......@@ -64,8 +64,8 @@ def lal_cache(urls):
class OmegaCacheEntry(namedtuple(
'OmegaCacheEntry',
('obs', 'tag', 'segment', 'duration', 'url')
"OmegaCacheEntry",
("obs", "tag", "segment", "duration", "url")
)):
"""CacheEntry for an omega-style cache.
......@@ -74,7 +74,9 @@ class OmegaCacheEntry(namedtuple(
<obs> <tag> <dir-start> <dir-end> <file-duration> <directory>
"""
def __str__(self):
"""Return a `str` representation of this `OmegaCacheEntry`."""
return " ".join(map(str, (
self.obs,
self.tag,
......@@ -97,7 +99,7 @@ def omega_cache(cache):
wentry = None
for entry in sorted(
cache,
key=attrgetter('obs', 'tag', 'segment'),
key=attrgetter("obs", "tag", "segment"),
):
dir_ = os.path.dirname(entry.url)
......
......@@ -18,11 +18,11 @@
"""Tests for :mod:`gwdatafind`.
"""
__author__ = 'Duncan Macleod <duncan.macleod@ligo.org>'
__author__ = "Duncan Macleod <duncan.macleod@ligo.org>"
import pytest
if pytest.__version__ < '3.0':
if pytest.__version__ < "3.0":
yield_fixture = pytest.yield_fixture
else:
yield_fixture = pytest.fixture
......@@ -28,7 +28,7 @@ def tmpname():
"""Return a temporary file name, cleaning up after the method returns.
"""
name = tempfile.mktemp()
open(name, 'w').close()
open(name, "w").close()
try:
yield name
finally:
......
......@@ -25,7 +25,7 @@ import pytest
from .. import api
__author__ = 'Duncan Macleod <duncan.macleod@ligo.org>'
__author__ = "Duncan Macleod <duncan.macleod@ligo.org>"
def test_ping_path():
......
......@@ -28,15 +28,15 @@ from igwn_segments import segment
from .. import __main__ as main
__author__ = 'Duncan Macleod <duncan.macleod@ligo.org>'
__author__ = "Duncan Macleod <duncan.macleod@ligo.org>"
URLS = [
'file:///test/X-test-0-1.gwf',
'file:///test/X-test-1-1.gwf',
'file:///test2/X-test-2-1.gwf',
'file:///test2/X-test-7-4.gwf',
'file:///test/X-test-0-1.h5',
'file:///test/X-test-1-1.h5',
"file:///test/X-test-0-1.gwf",
"file:///test/X-test-1-1.gwf",
"file:///test2/X-test-2-1.gwf",
"file:///test2/X-test-7-4.gwf",
"file:///test/X-test-0-1.h5",
"file:///test/X-test-1-1.h5",
]
GWF_URLS = [url for url in URLS if url.endswith(".gwf")]
GWF_OUTPUT_URLS = """
......@@ -65,49 +65,49 @@ X test 7 11 4 file:///test2
GAPS = [(3, 7)]
@mock.patch.dict(os.environ, {'GWDATAFIND_SERVER': 'something'})
@mock.patch.dict(os.environ, {"GWDATAFIND_SERVER": "something"})
def test_command_line():
parser = main.command_line()
assert isinstance(parser, argparse.ArgumentParser)
assert parser.description == main.__doc__
for query in ('ping', 'show_observatories', 'show_types', 'show_times',
'filename', 'latest'):
for query in ("ping", "show_observatories", "show_types", "show_times",
"filename", "latest"):
assert not parser.get_default(query)
assert parser.get_default('server') == os.getenv('GWDATAFIND_SERVER')
assert parser.get_default('url_type') == 'file'
assert parser.get_default('gaps') is False
assert parser.get_default("server") == os.getenv("GWDATAFIND_SERVER")
assert parser.get_default("url_type") == "file"
assert parser.get_default("gaps") is False
# test parsing and types
args = parser.parse_args([
'-o', 'X', '-t', 'test', '--gps-start-time', '0', '-e', '1',
"-o", "X", "-t", "test", "--gps-start-time", "0", "-e", "1",
])
assert args.gpsstart == 0.
assert args.gpsend == 1.
assert args.server == 'something'
assert args.server == "something"
@mock.patch.dict('os.environ', clear=True)
@pytest.mark.parametrize('defserv', (None, 'test.datafind.com:443'))
@mock.patch.dict("os.environ", clear=True)
@pytest.mark.parametrize("defserv", (None, "test.datafind.com:443"))
def test_command_line_server(defserv):
if defserv:
os.environ['GWDATAFIND_SERVER'] = defserv
os.environ["GWDATAFIND_SERVER"] = defserv
parser = main.command_line()
serveract = [act for act in parser._actions if act.dest == 'server'][0]
serveract = [act for act in parser._actions if act.dest == "server"][0]
assert serveract.required is (not defserv)
@mock.patch.dict(os.environ, {'GWDATAFIND_SERVER': 'something'})
@mock.patch.dict(os.environ, {"GWDATAFIND_SERVER": "something"})
def test_sanity_check_pass():
parser = main.command_line()
parser.parse_args(['-o', 'X', '-t', 'test', '-s', '0', '-e', '1'])
parser.parse_args(["-o", "X", "-t", "test", "-s", "0", "-e", "1"])
@mock.patch.dict(os.environ, {'GWDATAFIND_SERVER': 'something'})
@pytest.mark.parametrize('clargs', [
('--show-times', '--observatory', 'X'),
('--show-times', '--type', 'test'),
('--type', 'test', '--observatory', 'X', '--gps-start-time', '1'),
('--gaps', '--show-observatories'),
@mock.patch.dict(os.environ, {"GWDATAFIND_SERVER": "something"})
@pytest.mark.parametrize("clargs", [
("--show-times", "--observatory", "X"),
("--show-times", "--type", "test"),
("--type", "test", "--observatory", "X", "--gps-start-time", "1"),
("--gaps", "--show-observatories"),
])
def test_sanity_check_fail(clargs):
parser = main.command_line()
......@@ -115,11 +115,11 @@ def test_sanity_check_fail(clargs):
parser.parse_args(clargs)
@mock.patch('gwdatafind.ui.ping')
@mock.patch("gwdatafind.ui.ping")
def test_ping(mping):
args = argparse.Namespace(
server='test.datafind.com:443',
extension='gwf',
server="test.datafind.com:443",
extension="gwf",
)
out = StringIO()
main.ping(args, out)
......@@ -129,16 +129,16 @@ def test_ping(mping):
)
out.seek(0)
assert out.read().rstrip() == (
'LDRDataFindServer at test.datafind.com:443 is alive')
"LDRDataFindServer at test.datafind.com:443 is alive")
@mock.patch('gwdatafind.ui.find_observatories')
@mock.patch("gwdatafind.ui.find_observatories")
def test_show_observatories(mfindobs):
mfindobs.return_value = ['A', 'B', 'C']
mfindobs.return_value = ["A", "B", "C"]
args = argparse.Namespace(
server='test.datafind.com:443',
extension='gwf',
match='test',
server="test.datafind.com:443",
extension="gwf",
match="test",
)
out = StringIO()
main.show_observatories(args, out)
......@@ -148,17 +148,17 @@ def test_show_observatories(mfindobs):
match=args.match,
ext=args.extension,
)
assert list(map(str.rstrip, out.readlines())) == ['A', 'B', 'C']
assert list(map(str.rstrip, out.readlines())) == ["A", "B", "C"]
@mock.patch('gwdatafind.ui.find_types')
@mock.patch("gwdatafind.ui.find_types")
def test_show_types(mfindtypes):
mfindtypes.return_value = ['A', 'B', 'C']
mfindtypes.return_value = ["A", "B", "C"]
args = argparse.Namespace(
server='test.datafind.com:443',
extension='gwf',
observatory='X',
match='test',
server="test.datafind.com:443",
extension="gwf",
observatory="X",
match="test",
)
out = StringIO()
main.show_types(args, out)
......@@ -169,17 +169,17 @@ def test_show_types(mfindtypes):
site=args.observatory,
ext=args.extension,
)
assert list(map(str.rstrip, out.readlines())) == ['A', 'B', 'C']
assert list(map(str.rstrip, out.readlines())) == ["A", "B", "C"]
@mock.patch('gwdatafind.ui.find_times')
@mock.patch("gwdatafind.ui.find_times")
def test_show_times(mfindtimes):
mfindtimes.return_value = [segment(0, 1), segment(1, 2), segment(3, 4)]
args = argparse.Namespace(
server='test.datafind.com:443',
extension='gwf',
observatory='X',
type='test',
server="test.datafind.com:443",
extension="gwf",
observatory="X",
type="test",
gpsstart=0,
gpsend=10,
)
......@@ -199,15 +199,15 @@ def test_show_times(mfindtimes):
assert line.split() == list(map(str, (i, seg[0], seg[1], abs(seg))))
@mock.patch('gwdatafind.ui.find_latest')
@mock.patch("gwdatafind.ui.find_latest")
def test_latest(mlatest):
mlatest.return_value = ['file:///test/X-test-0-10.gwf']
mlatest.return_value = ["file:///test/X-test-0-10.gwf"]
args = argparse.Namespace(
server='test.datafind.com:443',
extension='gwf',
observatory='X',
type='test',
url_type='file',
server="test.datafind.com:443",
extension="gwf",
observatory="X",
type="test",
url_type="file",
format="urls",
gaps=None,
)
......@@ -217,7 +217,7 @@ def test_latest(mlatest):
args.observatory,
args.type,
urltype=args.url_type,
on_missing='warn',
on_missing="warn",
host=args.server,
ext=args.extension,
)
......@@ -225,13 +225,13 @@ def test_latest(mlatest):
assert out.read().rstrip() == mlatest.return_value[0]
@mock.patch('gwdatafind.ui.find_url')
@mock.patch("gwdatafind.ui.find_url")
def test_filename(mfindurl):
mfindurl.return_value = ['file:///test/X-test-0-10.gwf']
mfindurl.return_value = ["file:///test/X-test-0-10.gwf"]
args = argparse.Namespace(
server='test.datafind.com:443',
filename='X-test-0-10.gwf',
url_type='file',
server="test.datafind.com:443",
filename="X-test-0-10.gwf",
url_type="file",
type=None,
format="urls",
gaps=None,
......@@ -241,14 +241,14 @@ def test_filename(mfindurl):
mfindurl.assert_called_with(
args.filename,
urltype=args.url_type,
on_missing='warn',
on_missing="warn",
host=args.server,
)
out.seek(0)
assert out.read().rstrip() == mfindurl.return_value[0]
@mock.patch('gwdatafind.ui.find_urls')
@mock.patch("gwdatafind.ui.find_urls")
@pytest.mark.parametrize("ext", [
"gwf",
"h5",
......@@ -257,13 +257,13 @@ def test_show_urls(mfindurls, ext):
urls = [x for x in URLS if x.endswith(f".{ext}")]
mfindurls.return_value = urls
args = argparse.Namespace(
server='test.datafind.com:443',
server="test.datafind.com:443",
extension=ext,
observatory='X',
type='test',
observatory="X",
type="test",
gpsstart=0,
gpsend=10,
url_type='file',
url_type="file",
match=None,
format="urls",
gaps=None,
......@@ -277,7 +277,7 @@ def test_show_urls(mfindurls, ext):
args.gpsend,
match=args.match,
urltype=args.url_type,
on_gaps='ignore',
on_gaps="ignore",
ext=ext,
host=args.server,
)
......@@ -285,7 +285,7 @@ def test_show_urls(mfindurls, ext):
assert list(map(str.rstrip, out.readlines())) == urls
@pytest.mark.parametrize('fmt,result', [
@pytest.mark.parametrize("fmt,result", [
("urls", GWF_OUTPUT_URLS),
("lal", GWF_OUTPUT_LAL_CACHE),
("names", GWF_OUTPUT_NAMES_ONLY),
......@@ -310,14 +310,14 @@ def test_postprocess_cache_format(fmt, result):
def test_postprocess_cache_sft():
args = argparse.Namespace(
type='TEST_1800SFT',
type="TEST_1800SFT",
format=None,
gaps=None,
)
out = StringIO()
main.postprocess_cache(GWF_URLS, args, out)
out.seek(0)
assert out.read() == GWF_OUTPUT_URLS.replace('.gwf', '.sft')
assert out.read() == GWF_OUTPUT_URLS.replace(".gwf", ".sft")
def test_postprocess_cache_gaps(capsys):
......@@ -339,22 +339,22 @@ def test_postprocess_cache_gaps(capsys):
assert main.postprocess_cache(URLS, args, out) == 2
@mock.patch.dict(os.environ, {'GWDATAFIND_SERVER': 'something'})
@pytest.mark.parametrize('args,patch', [
(['--ping'], 'ping'),
(['--show-observatories'], 'show_observatories'),
(['--show-types'], 'show_types'),
(['--show-times', '-o', 'X', '-t', 'test'], 'show_times'),
(['--latest', '-o', 'X', '-t', 'test'], 'latest'),
(['--filename', 'X-test-0-1.gwf'], 'filename'),
(['-o', 'X', '-t', 'test', '-s', '0', '-e', '10'], 'show_urls'),
@mock.patch.dict(os.environ, {"GWDATAFIND_SERVER": "something"})
@pytest.mark.parametrize("args,patch", [
(["--ping"], "ping"),
(["--show-observatories"], "show_observatories"),
(["--show-types"], "show_types"),
(["--show-times", "-o", "X", "-t", "test"], "show_times"),
(["--latest", "-o", "X", "-t", "test"], "latest"),
(["--filename", "X-test-0-1.gwf"], "filename"),
(["-o", "X", "-t", "test", "-s", "0", "-e", "10"], "show_urls"),
])
def test_main(args, patch, tmpname):
with mock.patch(f"gwdatafind.__main__.{patch}") as mocked:
main.main(args)
assert mocked.call_count == 1
# call again with output file
args.extend(('--output-file', tmpname))
args.extend(("--output-file", tmpname))
with mock.patch(f"gwdatafind.__main__.{patch}") as mocked:
main.main(args)
assert mocked.call_count == 1
......@@ -28,7 +28,7 @@ from .. import (
)
from . import yield_fixture
__author__ = 'Duncan Macleod <duncan.macleod@ligo.org>'
__author__ = "Duncan Macleod <duncan.macleod@ligo.org>"
TEST_SERVER = "test.datafind.org"
TEST_URL_BASE = f"https://{TEST_SERVER}"
......
......@@ -37,7 +37,7 @@ from .utils import (
get_default_host,
)
__author__ = 'Duncan Macleod <duncan.macleod@ligo.org>'
__author__ = "Duncan Macleod <duncan.macleod@ligo.org>"
__all__ = [
"ping",
......@@ -80,7 +80,7 @@ def get_json(*args, **kwargs):
data : `object`
the URL reponse parsed with :func:`json.loads`
See also
See Also
--------
igwn_auth_utils.requests.get
for information on how the request is performed
......
......@@ -16,8 +16,7 @@
# with this program; if not, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
"""Utilities for the GW datafind service.
"""
"""Utilities for the GW datafind service."""
import os
import warnings
......@@ -153,9 +152,9 @@ def filename_metadata(filename):
file naming convention that includes documenting the GPS start integer
and integer duration of a file, see that document for more details.
"""
obs, desc, start, end = os.path.basename(filename).split('-')
obs, desc, start, end = os.path.basename(filename).split("-")
start = int(start)
end = int(end.split('.')[0])
end = int(end.split(".")[0])
return obs, desc, segment(start, start+end)
......
......@@ -80,11 +80,24 @@ source = [
]
[tool.coverage.report]
exclude_lines = [
# ignore when asked
"pragma: no( |-)cover",
# don't complain about typing blocks
"if (typing\\.)?TYPE_CHECKING",
]
precision = 1
[tool.coverage.run]
source = ["gwdatafind"]
[tool.mypy]
check_untyped_defs = true
exclude = [
"^docs/",
]
ignore_missing_imports = true
[tool.pytest.ini_options]
addopts = "-r a --color=yes"
filterwarnings = [
......@@ -93,23 +106,14 @@ filterwarnings = [
]
[tool.ruff.lint]
select = [
# mccabe
"C90",
# pycodestyle errors
"E",
# flake8-executable
"EXE",
# pyflakes
"F",
# isort
"I",
# pep8-naming
"N",
# pyupgrade
"UP",
# pycodestyle warnings
"W",
select = ["ALL"]
ignore = [
"ANN003", # type annotations for **kwargs
"D203", # blank line before class docstring
"D213", # docstring summary on second line
"D413", # blank line after last section
"PLR0913", # too many arguments
"SIM108", # if-else instead of ternary if
]
[tool.ruff.lint.isort]
......@@ -121,6 +125,19 @@ force-wrap-aliases = true
"F401", # unused-import
"F403", # undefined-local-with-import-star
]
"*/tests/*" = [
"ANN", # type annotations
"B904", # raise from
"EM101", # string literal in exception
"PLR2004", # magic value used in comparison
"S101", # assert
]
"docs/*" = [
"A", # builtins
"ANN", # type annotations
"D", # docstrings
"INP001", # implicit namespace package
]
[tool.setuptools]
license-files = [ "LICENSE" ]
......