<commit_before># Licensed under a 3-clause BSD style license - see LICENSE.rst
from contextlib import contextmanager
import os
import socket
import numpy as np
from numpy.testing import assert_allclose

import six
from astropy.io import fits
from astropy.table import Column, Table
import pytest

from ... import sdss
from ...utils.testing_tools import MockResponse
from ...exceptions import TimeoutError
from ...utils import commons

# actual spectra/data are a bit heavy to include in astroquery, so we don't try
# to deal with them.  Would be nice to find a few very small examples

DATA_FILES = {'spectra_id': 'xid_sp.txt',
              'images_id': 'xid_im.txt',
              'spectra': 'emptyfile.fits',
              'images': 'emptyfile.fits'}


@pytest.fixture
def patch_get(request):
    try:
        mp = request.getfixturevalue("monkeypatch")
    except AttributeError:  # pytest < 3
        mp = request.getfuncargvalue("monkeypatch")
    mp.setattr(sdss.SDSS, '_request', get_mockreturn)
    return mp


@pytest.fixture
def patch_post(request):
    try:
        mp = request.getfixturevalue("monkeypatch")
    except AttributeError:  # pytest < 3
        mp = request.getfuncargvalue("monkeypatch")
    mp.setattr(sdss.SDSS, '_request', post_mockreturn)
    return mp


@pytest.fixture
def patch_get_slow(request):
    try:
        mp = request.getfixturevalue("monkeypatch")
    except AttributeError:  # pytest < 3
        mp = request.getfuncargvalue("monkeypatch")
    mp.setattr(sdss.SDSS, '_request', get_mockreturn_slow)
    return mp


@pytest.fixture
def patch_get_readable_fileobj(request):
    @contextmanager
    def get_readable_fileobj_mockreturn(filename, **kwargs):
        file_obj = data_path(DATA_FILES['spectra'])  # TODO: add images option
        encoding = kwargs.get('encoding', None)
        if encoding == 'binary':
            yield open(file_obj, 'rb')
        else:
            yield open(file_obj, 'r', encoding=encoding)

    try:
        mp = request.getfixturevalue("monkeypatch")
    except AttributeError:  # pytest < 3
        mp = request.getfuncargvalue("monkeypatch")
    mp.setattr(commons, 'get_readable_fileobj',
               get_readable_fileobj_mockreturn)
    return mp


@pytest.fixture
def patch_get_readable_fileobj_slow(request):
    @contextmanager
    def get_readable_fileobj_mockreturn(filename, **kwargs):
        e = six.moves.urllib_error.URLError('timeout')
        e.reason = socket.timeout()
        raise e
        yield True
    try:
        mp = request.getfixturevalue("monkeypatch")
    except AttributeError:  # pytest < 3
        mp = request.getfuncargvalue("monkeypatch")
    mp.setattr(commons, 'get_readable_fileobj',
               get_readable_fileobj_mockreturn)
    return mp


def get_mockreturn(method, url, params=None, timeout=10, cache=True, **kwargs):
    if 'SpecObjAll' in params['cmd']:
        filename = data_path(DATA_FILES['spectra_id'])
    else:
        filename = data_path(DATA_FILES['images_id'])
    content = open(filename, 'rb').read()
    return MockResponse(content, **kwargs)


def get_mockreturn_slow(method, url, params=None, timeout=0, **kwargs):
    raise TimeoutError


def post_mockreturn(method, url, params=None, timeout=0, **kwargs):
    filename = data_path(DATA_FILES['images_id'])
    content = open(filename, 'rb').read()
    return MockResponse(content)


def data_path(filename):
    data_dir = os.path.join(os.path.dirname(__file__), 'data')
    return os.path.join(data_dir, filename)


# Test Case: A Seyfert 1 galaxy
coords = commons.ICRSCoordGenerator('0h8m05.63s +14d50m23.3s')

# Test Case: list of coordinates
coords_list = [coords, coords]

# Test Case: Column of coordinates
coords_column = Column(coords_list, name='coordinates')

# List of all data releases.
dr_list = list(range(1, sdss.conf.default_release + 1))

# We are not testing queries for DR11 because it is not easily available to
# query: "DR11 data are distributed primarily to provide reproducibility of
# published results based on the DR11 data set. As such, not all data-access
# interfaces are supported for DR11."
def url_tester(data_release):
    if data_release < 10:
        baseurl = 'http://skyserver.sdss.org/dr{}/en/tools/search/x_sql.asp'
    if data_release == 10:
        baseurl = 'http://skyserver.sdss.org/dr{}/en/tools/search/x_sql.aspx'
    if data_release == 11:<commit_msg>fix pycodestyle error<commit_after># Licensed under a 3-clause BSD style license - see LICENSE.rst
from contextlib import contextmanager
import os
import socket
import numpy as np
from numpy.testing import assert_allclose

import six
from astropy.io import fits
from astropy.table import Column, Table
import pytest

from ... import sdss
from ...utils.testing_tools import MockResponse
from ...exceptions import TimeoutError
from ...utils import commons

# actual spectra/data are a bit heavy to include in astroquery, so we don't try
# to deal with them.  Would be nice to find a few very small examples

DATA_FILES = {'spectra_id': 'xid_sp.txt',
              'images_id': 'xid_im.txt',
              'spectra': 'emptyfile.fits',
              'images': 'emptyfile.fits'}


@pytest.fixture
def patch_get(request):
    try:
        mp = request.getfixturevalue("monkeypatch")
    except AttributeError:  # pytest < 3
        mp = request.getfuncargvalue("monkeypatch")
    mp.setattr(sdss.SDSS, '_request', get_mockreturn)
    return mp


@pytest.fixture
def patch_post(request):
    try:
        mp = request.getfixturevalue("monkeypatch")
    except AttributeError:  # pytest < 3
        mp = request.getfuncargvalue("monkeypatch")
    mp.setattr(sdss.SDSS, '_request', post_mockreturn)
    return mp


@pytest.fixture
def patch_get_slow(request):
    try:
        mp = request.getfixturevalue("monkeypatch")
    except AttributeError:  # pytest < 3
        mp = request.getfuncargvalue("monkeypatch")
    mp.setattr(sdss.SDSS, '_request', get_mockreturn_slow)
    return mp


@pytest.fixture
def patch_get_readable_fileobj(request):
    @contextmanager
    def get_readable_fileobj_mockreturn(filename, **kwargs):
        file_obj = data_path(DATA_FILES['spectra'])  # TODO: add images option
        encoding = kwargs.get('encoding', None)
        if encoding == 'binary':
            yield open(file_obj, 'rb')
        else:
            yield open(file_obj, 'r', encoding=encoding)

    try:
        mp = request.getfixturevalue("monkeypatch")
    except AttributeError:  # pytest < 3
        mp = request.getfuncargvalue("monkeypatch")
    mp.setattr(commons, 'get_readable_fileobj',
               get_readable_fileobj_mockreturn)
    return mp


@pytest.fixture
def patch_get_readable_fileobj_slow(request):
    @contextmanager
    def get_readable_fileobj_mockreturn(filename, **kwargs):
        e = six.moves.urllib_error.URLError('timeout')
        e.reason = socket.timeout()
        raise e
        yield True
    try:
        mp = request.getfixturevalue("monkeypatch")
    except AttributeError:  # pytest < 3
        mp = request.getfuncargvalue("monkeypatch")
    mp.setattr(commons, 'get_readable_fileobj',
               get_readable_fileobj_mockreturn)
    return mp


def get_mockreturn(method, url, params=None, timeout=10, cache=True, **kwargs):
    if 'SpecObjAll' in params['cmd']:
        filename = data_path(DATA_FILES['spectra_id'])
    else:
        filename = data_path(DATA_FILES['images_id'])
    content = open(filename, 'rb').read()
    return MockResponse(content, **kwargs)


def get_mockreturn_slow(method, url, params=None, timeout=0, **kwargs):
    raise TimeoutError


def post_mockreturn(method, url, params=None, timeout=0, **kwargs):
    filename = data_path(DATA_FILES['images_id'])
    content = open(filename, 'rb').read()
    return MockResponse(content)


def data_path(filename):
    data_dir = os.path.join(os.path.dirname(__file__), 'data')
    return os.path.join(data_dir, filename)


# Test Case: A Seyfert 1 galaxy
coords = commons.ICRSCoordGenerator('0h8m05.63s +14d50m23.3s')

# Test Case: list of coordinates
coords_list = [coords, coords]

# Test Case: Column of coordinates
coords_column = Column(coords_list, name='coordinates')

# List of all data releases.
dr_list = list(range(1, sdss.conf.default_release + 1))


# We are not testing queries for DR11 because it is not easily available to
# query: "DR11 data are distributed primarily to provide reproducibility of
# published results based on the DR11 data set. As such, not all data-access
# interfaces are supported for DR11."
def url_tester(data_release):
    if data_release < 10:
        baseurl = 'http://skyserver.sdss.org/dr{}/en/tools/search/x_sql.asp'
    if data_release == 10:
        baseurl = 'http://skyserver.sdss.org/dr{}/en/tools/search/x_sql.aspx'
    if data_release == 11:
{'fraction_non_alphanumeric': 0.10529899012821968, 'fraction_numerical': 0.009077499148984454, 'mean_word_length': 3.605015673981191, 'pattern_counts': {'":': 0, '<': 15, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7569189', 'n_tokens_mistral': 2884, 'n_tokens_neox': 2762, 'n_words': 709}