# Path: pypi_portal/extensions.py
# LOG = getLogger(__name__)
# def _on_connect(dbapi_connection, **_):
# 
# Path: pypi_portal/models/pypi.py
# class Package(Base):
#     """Holds all of the packages on PyPI."""
#     name = Column(String(255), unique=True, nullable=False)
#     summary = Column(Text, nullable=False)
#     latest_version = Column(String(128), nullable=False)
# 
# Path: pypi_portal/models/redis.py
# POLL_SIMPLE_THROTTLE = 'pypi_portal:poll_simple_throttle'  # Lock.
# 
# Path: pypi_portal/tasks/pypi.py
# LOG = getLogger(__name__)
# THROTTLE = 1 * 60 * 60
# def update_package_list():


from flask import current_app
from flask.ext.celery import CELERY_LOCK
from redis.exceptions import LockError
from pypi_portal.extensions import db, redis
from pypi_portal.models.pypi import Package
from pypi_portal.models.redis import POLL_SIMPLE_THROTTLE
from pypi_portal.tasks import pypi
import pytest
    actual = db.session.query(Package.name, Package.summary, Package.latest_version).all()
    assert sorted(expected) == sorted(actual)


def test_sync_unhandled_exception():
    old_throttle = pypi.THROTTLE
    pypi.THROTTLE = 'nan'
    redis.delete(POLL_SIMPLE_THROTTLE)

    with pytest.raises(ValueError):
        current_app.test_client().get('/pypi/sync').status()

    pypi.THROTTLE = old_throttle


def test_sync_timeout():
    old_delay = pypi.update_package_list.delay
    pypi.update_package_list.delay = FakeDelay
    redis.delete(POLL_SIMPLE_THROTTLE)

    assert '302 FOUND' == current_app.test_client().get('/pypi/sync').status

    expected = [
        ('packageB', 'Test package.', '3.0.0'), ('packageB1', 'Test package.', '3.0.0'),
        ('packageB2', 'Test package.', '3.0.0'), ('packageB3', 'Test package.', '3.0.0'),
        ('packageB4', 'Test package.', '3.0.0'), ('packageB5', 'Test package.', '3.0.0'),
    ]
    actual = db.session.query(Package.name, Package.summary, Package.latest_version).all()
    assert sorted(expected) == sorted(actual)


# Path: pypi_portal/extensions.py
# LOG = getLogger(__name__)
# def _on_connect(dbapi_connection, **_):
# 
# Path: pypi_portal/models/pypi.py
# class Package(Base):
#     """Holds all of the packages on PyPI."""
#     name = Column(String(255), unique=True, nullable=False)
#     summary = Column(Text, nullable=False)
#     latest_version = Column(String(128), nullable=False)
# 
# Path: pypi_portal/models/redis.py
# POLL_SIMPLE_THROTTLE = 'pypi_portal:poll_simple_throttle'  # Lock.
# 
# Path: pypi_portal/tasks/pypi.py
# LOG = getLogger(__name__)
# THROTTLE = 1 * 60 * 60
# def update_package_list():


# Path: tests/views/test_pypi_packages.py
from flask import current_app
from flask.ext.celery import CELERY_LOCK
from redis.exceptions import LockError
from pypi_portal.extensions import db, redis
from pypi_portal.models.pypi import Package
from pypi_portal.models.redis import POLL_SIMPLE_THROTTLE
from pypi_portal.tasks import pypi
import pytest

    actual = db.session.query(Package.name, Package.summary, Package.latest_version).all()
    assert sorted(expected) == sorted(actual)


def test_sync_unhandled_exception():
    old_throttle = pypi.THROTTLE
    pypi.THROTTLE = 'nan'
    redis.delete(POLL_SIMPLE_THROTTLE)

    with pytest.raises(ValueError):
        current_app.test_client().get('/pypi/sync').status()

    pypi.THROTTLE = old_throttle


def test_sync_timeout():
    old_delay = pypi.update_package_list.delay
    pypi.update_package_list.delay = FakeDelay
    redis.delete(POLL_SIMPLE_THROTTLE)

    assert '302 FOUND' == current_app.test_client().get('/pypi/sync').status

    expected = [
        ('packageB', 'Test package.', '3.0.0'), ('packageB1', 'Test package.', '3.0.0'),
        ('packageB2', 'Test package.', '3.0.0'), ('packageB3', 'Test package.', '3.0.0'),
        ('packageB4', 'Test package.', '3.0.0'), ('packageB5', 'Test package.', '3.0.0'),
    ]
    actual = db.session.query(Package.name, Package.summary, Package.latest_version).all()
    assert sorted(expected) == sorted(actual)


    pypi.update_package_list.delay = old_delay