external/aurweb
0
0
Fork 0
mirror of https://gitlab.archlinux.org/archlinux/aurweb.git synced 2025-02-03 10:43:03 +01:00
Code Issues Projects Releases Packages Wiki Activity Actions

housekeep: centralize datetime generation

Browse source 
Signed-off-by: Kevin Morris <kevr@0cost.org>
This commit is contained in:
Kevin Morris 2022-01-18 05:59:52 -08:00
parent 7bcc8d7ce7
commit 7f6c23d4cb
No known key found for this signature in database
GPG key ID: F7E46DED420788F3
41 changed files with 146 additions and 179 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

5
aurweb/auth/__init__.py
View file

@ -1,6 +1,5 @@
import functools
from datetime import datetime
from http import HTTPStatus
from typing import Callable
@ -13,7 +12,7 @@ from starlette.requests import HTTPConnection
import aurweb.config
from aurweb import db, filters, l10n, util
from aurweb import db, filters, l10n, time, util
from aurweb.models import Session, User
from aurweb.models.account_type import ACCOUNT_TYPE_ID
@ -110,7 +109,7 @@ class BasicAuthBackend(AuthenticationBackend):
"persistent_cookie_timeout")
# If no session with sid and a LastUpdateTS now or later exists.
now_ts = int(datetime.utcnow().timestamp())
now_ts = time.utcnow()
record = db.query(Session).filter(Session.SessionID == sid).first()
if not record:
return unauthenticated

6
aurweb/models/package_base.py
View file

@ -1,9 +1,7 @@
from datetime import datetime
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import backref, relationship
from aurweb import schema
from aurweb import schema, time
from aurweb.models.declarative import Base
from aurweb.models.user import User as _User
@ -43,7 +41,7 @@ class PackageBase(Base):
# If no SubmittedTS/ModifiedTS is provided on creation, set them
# here to the current utc timestamp.
now = datetime.utcnow().timestamp()
now = time.utcnow()
if not self.SubmittedTS:
self.SubmittedTS = now
if not self.ModifiedTS:

6
aurweb/models/tu_voteinfo.py
View file

@ -1,11 +1,9 @@
import typing
from datetime import datetime
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import backref, relationship
from aurweb import schema
from aurweb import schema, time
from aurweb.models.declarative import Base
from aurweb.models.user import User as _User
@ -71,7 +69,7 @@ class TUVoteInfo(Base):
return attr
def is_running(self):
return self.End > int(datetime.utcnow().timestamp())
return self.End > time.utcnow()
def total_votes(self):
return self.Yes + self.No + self.Abstain

5
aurweb/models/user.py
View file

@ -1,6 +1,5 @@
import hashlib
from datetime import datetime
from typing import List, Set
import bcrypt
@ -14,7 +13,7 @@ import aurweb.config
import aurweb.models.account_type
import aurweb.schema
from aurweb import db, logging, schema, util
from aurweb import db, logging, schema, time, util
from aurweb.models.account_type import AccountType as _AccountType
from aurweb.models.ban import is_banned
from aurweb.models.declarative import Base
@ -121,7 +120,7 @@ class User(Base):
exc = None
for i in range(tries):
exc = None
now_ts = datetime.utcnow().timestamp()
now_ts = time.utcnow()
try:
with db.begin():
self.LastLogin = now_ts

7
aurweb/packages/requests.py
View file

@ -1,10 +1,9 @@
from datetime import datetime
from typing import List, Optional, Set
from fastapi import Request
from sqlalchemy import and_, orm
from aurweb import config, db, l10n, util
from aurweb import config, db, l10n, time, util
from aurweb.exceptions import InvariantError
from aurweb.models import PackageBase, PackageRequest, User
from aurweb.models.package_request import ACCEPTED_ID, PENDING_ID, REJECTED_ID
@ -106,7 +105,7 @@ def verify_orphan_request(user: User, pkgbase: PackageBase):
PackageRequest.ReqTypeID == ORPHAN_ID)
for pkgreq in requests:
idle_time = config.getint("options", "request_idle_time")
time_delta = int(datetime.utcnow().timestamp()) - pkgreq.RequestTS
time_delta = time.utcnow() - pkgreq.RequestTS
is_due = pkgreq.Status == PENDING_ID and time_delta > idle_time
if is_due:
# If the requester is the pkgbase maintainer or the
@ -128,7 +127,7 @@ def close_pkgreq(pkgreq: PackageRequest, closer: User,
:param target: Optional PackageBase instance to merge into
:param status: `pkgreq`.Status value to update to
"""
now = int(datetime.utcnow().timestamp())
now = time.utcnow()
pkgreq.Status = status
pkgreq.Closer = closer
pkgreq.ClosureComment = (

8
aurweb/ratelimit.py
View file

@ -1,9 +1,7 @@
from datetime import datetime
from fastapi import Request
from redis.client import Pipeline
from aurweb import config, db, logging
from aurweb import config, db, logging, time
from aurweb.models import ApiRateLimit
from aurweb.redis import redis_connection
@ -12,7 +10,7 @@ logger = logging.get_logger(__name__)
def _update_ratelimit_redis(request: Request, pipeline: Pipeline):
window_length = config.getint("ratelimit", "window_length")
now = int(datetime.utcnow().timestamp())
now = time.utcnow()
time_to_delete = now - window_length
host = request.client.host
@ -37,7 +35,7 @@ def _update_ratelimit_redis(request: Request, pipeline: Pipeline):
def _update_ratelimit_db(request: Request):
window_length = config.getint("ratelimit", "window_length")
now = int(datetime.utcnow().timestamp())
now = time.utcnow()
time_to_delete = now - window_length
records = db.query(ApiRateLimit).filter(

6
aurweb/routers/auth.py
View file

@ -1,4 +1,3 @@
from datetime import datetime
from http import HTTPStatus
from fastapi import APIRouter, Form, HTTPException, Request
@ -6,7 +5,7 @@ from fastapi.responses import HTMLResponse, RedirectResponse
import aurweb.config
from aurweb import cookies, db
from aurweb import cookies, db, time
from aurweb.auth import requires_auth, requires_guest
from aurweb.l10n import get_translator_for_request
from aurweb.models import User
@ -57,8 +56,7 @@ async def login_post(request: Request,
login_timeout = aurweb.config.getint("options", "login_timeout")
expires_at = int(datetime.utcnow().timestamp()
+ max(cookie_timeout, login_timeout))
expires_at = int(time.utcnow() + max(cookie_timeout, login_timeout))
response = RedirectResponse(url=next,
status_code=HTTPStatus.SEE_OTHER)

5
aurweb/routers/html.py
View file

@ -3,7 +3,6 @@ decorators in some way; more complex routes should be defined in their
own modules and imported here. """
import os
from datetime import datetime
from http import HTTPStatus
from fastapi import APIRouter, Form, HTTPException, Request, Response
@ -14,7 +13,7 @@ from sqlalchemy import and_, case, or_
import aurweb.config
import aurweb.models.package_request
from aurweb import cookies, db, models, util
from aurweb import cookies, db, models, time, util
from aurweb.cache import db_count_cache
from aurweb.models.account_type import TRUSTED_USER_AND_DEV_ID, TRUSTED_USER_ID
from aurweb.models.package_request import PENDING_ID
@ -97,7 +96,7 @@ async def index(request: Request):
redis, "trusted_user_count", query, expire=cache_expire)
# Current timestamp.
now = int(datetime.utcnow().timestamp())
now = time.utcnow()
seven_days = 86400 * 7 # Seven days worth of seconds.
seven_days_ago = now - seven_days

17
aurweb/routers/pkgbase.py
View file

@ -1,11 +1,10 @@
from datetime import datetime
from http import HTTPStatus
from fastapi import APIRouter, Form, HTTPException, Query, Request, Response
from fastapi.responses import JSONResponse, RedirectResponse
from sqlalchemy import and_
from aurweb import config, db, l10n, logging, templates, util
from aurweb import config, db, l10n, logging, templates, time, util
from aurweb.auth import creds, requires_auth
from aurweb.exceptions import InvariantError, ValidationError
from aurweb.models import PackageBase
@ -146,7 +145,7 @@ async def pkgbase_flag_post(request: Request, name: str,
has_cred = request.user.has_credential(creds.PKGBASE_FLAG)
if has_cred and not pkgbase.Flagger:
now = int(datetime.utcnow().timestamp())
now = time.utcnow()
with db.begin():
pkgbase.OutOfDateTS = now
pkgbase.Flagger = request.user
@ -170,7 +169,7 @@ async def pkgbase_comments_post(
# If the provided comment is different than the record's version,
# update the db record.
now = int(datetime.utcnow().timestamp())
now = time.utcnow()
with db.begin():
comment = db.create(PackageComment, User=request.user,
PackageBase=pkgbase,
@ -268,7 +267,7 @@ async def pkgbase_comment_post(
# If the provided comment is different than the record's version,
# update the db record.
now = int(datetime.utcnow().timestamp())
now = time.utcnow()
if db_comment.Comments != comment:
with db.begin():
db_comment.Comments = comment
@ -316,7 +315,7 @@ async def pkgbase_comment_pin(request: Request, name: str, id: int,
status_code=HTTPStatus.UNAUTHORIZED,
detail=_("You are not allowed to pin this comment."))
now = int(datetime.utcnow().timestamp())
now = time.utcnow()
with db.begin():
comment.PinnedTS = now
@ -387,7 +386,7 @@ async def pkgbase_comment_delete(request: Request, name: str, id: int,
status_code=HTTPStatus.UNAUTHORIZED,
detail=_("You are not allowed to delete this comment."))
now = int(datetime.utcnow().timestamp())
now = time.utcnow()
with db.begin():
comment.Deleter = request.user
comment.DelTS = now
@ -446,7 +445,7 @@ async def pkgbase_vote(request: Request, name: str):
).first()
has_cred = request.user.has_credential(creds.PKGBASE_VOTE)
if has_cred and not vote:
now = int(datetime.utcnow().timestamp())
now = time.utcnow()
with db.begin():
db.create(PackageVote,
User=request.user,
@ -683,7 +682,7 @@ async def pkgbase_request_post(request: Request, name: str,
return render_template(request, "pkgbase/request.html", context)
# All good. Create a new PackageRequest based on the given type.
now = int(datetime.utcnow().timestamp())
now = time.utcnow()
with db.begin():
pkgreq = db.create(PackageRequest,
ReqTypeID=types.get(type),

5
aurweb/routers/requests.py
View file

@ -1,11 +1,10 @@
from datetime import datetime
from http import HTTPStatus
from fastapi import APIRouter, Form, Query, Request
from fastapi.responses import RedirectResponse
from sqlalchemy import case
from aurweb import db, defaults, util
from aurweb import db, defaults, time, util
from aurweb.auth import creds, requires_auth
from aurweb.models import PackageRequest, User
from aurweb.models.package_request import PENDING_ID, REJECTED_ID
@ -78,7 +77,7 @@ async def request_close_post(request: Request, id: int,
context = make_context(request, "Close Request")
context["pkgreq"] = pkgreq
now = int(datetime.utcnow().timestamp())
now = time.utcnow()
with db.begin():
pkgreq.Closer = request.user
pkgreq.ClosureComment = comments

7
aurweb/routers/trusted_user.py
View file

@ -2,14 +2,13 @@ import html
import re
import typing
from datetime import datetime
from http import HTTPStatus
from fastapi import APIRouter, Form, HTTPException, Request
from fastapi.responses import RedirectResponse, Response
from sqlalchemy import and_, or_
from aurweb import db, l10n, logging, models
from aurweb import db, l10n, logging, models, time
from aurweb.auth import creds, requires_auth
from aurweb.models import User
from aurweb.models.account_type import TRUSTED_USER_AND_DEV_ID, TRUSTED_USER_ID
@ -52,7 +51,7 @@ async def trusted_user(request: Request,
context["pp"] = pp = ITEMS_PER_PAGE
context["prev_len"] = MAX_AGENDA_LENGTH
ts = int(datetime.utcnow().timestamp())
ts = time.utcnow()
if current_by not in {"asc", "desc"}:
# If a malicious by was given, default to desc.
@ -290,7 +289,7 @@ async def trusted_user_addvote_post(request: Request,
# Gather some mapped constants and the current timestamp.
duration, quorum = ADDVOTE_SPECIFICS.get(type)
timestamp = int(datetime.utcnow().timestamp())
timestamp = time.utcnow()
# TODO: Review this. Is this even necessary?
# Remove <script> and <style> tags.

6
aurweb/scripts/pkgmaint.py
View file

@ -1,16 +1,14 @@
#!/usr/bin/env python3
from datetime import datetime
from sqlalchemy import and_
from aurweb import db
from aurweb import db, time
from aurweb.models import PackageBase
def _main():
# One day behind.
limit_to = int(datetime.utcnow().timestamp()) - 86400
limit_to = time.utcnow() - 86400
query = db.query(PackageBase).filter(
and_(PackageBase.SubmittedTS < limit_to,

5
aurweb/scripts/popupdate.py
View file

@ -1,13 +1,12 @@
#!/usr/bin/env python3
from datetime import datetime
from typing import List
from sqlalchemy import and_, func
from sqlalchemy.sql.functions import coalesce
from sqlalchemy.sql.functions import sum as _sum
from aurweb import db
from aurweb import db, time
from aurweb.models import PackageBase, PackageVote
@ -20,7 +19,7 @@ def run_variable(pkgbases: List[PackageBase] = []) -> None:
:param pkgbases: List of PackageBase instances
"""
now = int(datetime.utcnow().timestamp())
now = time.utcnow()
# NumVotes subquery.
votes_subq = db.get_session().query(

6
aurweb/scripts/tuvotereminder.py
View file

@ -1,12 +1,10 @@
#!/usr/bin/env python3
from datetime import datetime
from sqlalchemy import and_
import aurweb.config
from aurweb import db
from aurweb import db, time
from aurweb.models import TUVoteInfo
from aurweb.scripts import notify
@ -16,7 +14,7 @@ notify_cmd = aurweb.config.get('notifications', 'notify-cmd')
def main():
db.get_engine()
now = int(datetime.utcnow().timestamp())
now = time.utcnow()
start = aurweb.config.getint("tuvotereminder", "range_start")
filter_from = now + start

6
aurweb/scripts/usermaint.py
View file

@ -1,15 +1,13 @@
#!/usr/bin/env python3
from datetime import datetime
from sqlalchemy import update
from aurweb import db
from aurweb import db, time
from aurweb.models import User
def _main():
limit_to = int(datetime.utcnow().timestamp()) - 86400 * 7
limit_to = time.utcnow() - 86400 * 7
update_ = update(User).where(
User.LastLogin < limit_to

6
aurweb/templates.py
View file

@ -1,9 +1,7 @@
import copy
import functools
import os
import zoneinfo
from datetime import datetime
from http import HTTPStatus
from typing import Callable
@ -80,8 +78,8 @@ def make_context(request: Request, title: str, next: str = None):
"timezone": timezone,
"timezones": time.SUPPORTED_TIMEZONES,
"title": title,
"now": datetime.now(tz=zoneinfo.ZoneInfo(timezone)),
"utcnow": int(datetime.utcnow().timestamp()),
"now": time.now(timezone),
"utcnow": time.utcnow(),
"config": aurweb.config,
"creds": aurweb.auth.creds,
"next": next if next else request.url.path,

20
aurweb/time.py
View file

@ -3,6 +3,7 @@ import zoneinfo
from collections import OrderedDict
from datetime import datetime
from urllib.parse import unquote
from zoneinfo import ZoneInfo
from fastapi import Request
@ -62,3 +63,22 @@ def get_request_timezone(request: Request):
if request.user.is_authenticated():
default_tz = request.user.Timezone
return unquote(request.cookies.get("AURTZ", default_tz))
def now(timezone: str) -> datetime:
"""
Get the current timezone-localized timestamp.
:param timezone: Valid timezone supported by ZoneInfo
:return: Current localized datetime
"""
return datetime.now(tz=ZoneInfo(timezone))
def utcnow() -> int:
"""
Get the current UTC timestamp.
:return: Current UTC timestamp
"""
return int(datetime.utcnow().timestamp())

5
aurweb/users/update.py
View file

@ -1,9 +1,8 @@
from datetime import datetime
from typing import Any, Dict
from fastapi import Request
from aurweb import cookies, db, models
from aurweb import cookies, db, models, time
from aurweb.models.ssh_pub_key import get_fingerprint
from aurweb.util import strtobool
@ -14,7 +13,7 @@ def simple(U: str = str(), E: str = str(), H: bool = False,
CN: bool = False, UN: bool = False, ON: bool = False,
S: bool = False, user: models.User = None,
**kwargs) -> None:
now = int(datetime.utcnow().timestamp())
now = time.utcnow()
with db.begin():
user.Username = U or user.Username
user.Email = E or user.Email

4
test/test_accounts_routes.py
View file

@ -13,7 +13,7 @@ from fastapi.testclient import TestClient
import aurweb.models.account_type as at
from aurweb import captcha, db, logging
from aurweb import captcha, db, logging, time
from aurweb.asgi import app
from aurweb.db import create, query
from aurweb.models.accepted_term import AcceptedTerm
@ -155,7 +155,7 @@ def test_post_passreset_user(client: TestClient, user: User):
def test_post_passreset_resetkey(client: TestClient, user: User):
with db.begin():
user.session = Session(UsersID=user.ID, SessionID="blah",
LastUpdateTS=datetime.utcnow().timestamp())
LastUpdateTS=time.utcnow())
# Prepare a password reset.
with client as request:

10
test/test_auth.py
View file

@ -1,12 +1,10 @@
from datetime import datetime
import fastapi
import pytest
from fastapi import HTTPException
from sqlalchemy.exc import IntegrityError
from aurweb import config, db
from aurweb import config, db, time
from aurweb.auth import AnonymousUser, BasicAuthBackend, _auth_required, account_type_required
from aurweb.models.account_type import USER, USER_ID
from aurweb.models.session import Session
@ -55,7 +53,7 @@ async def test_auth_backend_invalid_sid(backend: BasicAuthBackend):
@pytest.mark.asyncio
async def test_auth_backend_invalid_user_id():
# Create a new session with a fake user id.
now_ts = datetime.utcnow().timestamp()
now_ts = time.utcnow()
with pytest.raises(IntegrityError):
Session(UsersID=666, SessionID="realSession",
LastUpdateTS=now_ts + 5)
@ -65,7 +63,7 @@ async def test_auth_backend_invalid_user_id():
async def test_basic_auth_backend(user: User, backend: BasicAuthBackend):
# This time, everything matches up. We expect the user to
# equal the real_user.
now_ts = datetime.utcnow().timestamp()
now_ts = time.utcnow()
with db.begin():
db.create(Session, UsersID=user.ID, SessionID="realSession",
LastUpdateTS=now_ts + 5)
@ -87,7 +85,7 @@ async def test_expired_session(backend: BasicAuthBackend, user: User):
# Set Session.LastUpdateTS to 20 seconds expired.
timeout = config.getint("options", "login_timeout")
now_ts = int(datetime.utcnow().timestamp())
now_ts = time.utcnow()
with db.begin():
request.user.session.LastUpdateTS = now_ts - timeout - 20

7
test/test_auth_routes.py
View file

@ -1,6 +1,5 @@
import re
from datetime import datetime
from http import HTTPStatus
from unittest import mock
@ -10,7 +9,7 @@ from fastapi.testclient import TestClient
import aurweb.config
from aurweb import db
from aurweb import db, time
from aurweb.asgi import app
from aurweb.models.account_type import USER_ID
from aurweb.models.session import Session
@ -200,7 +199,7 @@ def test_login_remember_me(client: TestClient, user: User):
cookie_timeout = aurweb.config.getint(
"options", "persistent_cookie_timeout")
now_ts = int(datetime.utcnow().timestamp())
now_ts = time.utcnow()
session = db.query(Session).filter(Session.UsersID == user.ID).first()
# Expect that LastUpdateTS is not past the cookie timeout
@ -294,7 +293,7 @@ def test_generate_unique_sid_exhausted(client: TestClient, user: User,
This exercises the bad path of /login, where we can't find a unique
SID to assign the user.
"""
now = int(datetime.utcnow().timestamp())
now = time.utcnow()
with db.begin():
# Create a second user; we'll login with this one.
user2 = db.create(User, Username="test2", Email="test2@example.org",

6
test/test_filters.py
View file

@ -1,17 +1,17 @@
from datetime import datetime
from zoneinfo import ZoneInfo
from aurweb import filters
from aurweb import filters, time
def test_timestamp_to_datetime():
ts = datetime.utcnow().timestamp()
ts = time.utcnow()
dt = datetime.utcfromtimestamp(int(ts))
assert filters.timestamp_to_datetime(ts) == dt
def test_as_timezone():
ts = datetime.utcnow().timestamp()
ts = time.utcnow()
dt = filters.timestamp_to_datetime(ts)
assert filters.as_timezone(dt, "UTC") == dt.astimezone(tz=ZoneInfo("UTC"))

9
test/test_homepage.py
View file

@ -1,6 +1,5 @@
import re
from datetime import datetime
from http import HTTPStatus
from unittest.mock import patch
@ -8,7 +7,7 @@ import pytest
from fastapi.testclient import TestClient
from aurweb import db
from aurweb import db, time
from aurweb.asgi import app
from aurweb.models.account_type import USER_ID
from aurweb.models.package import Package
@ -62,7 +61,7 @@ def packages(user):
# For i..num_packages, create a package named pkg_{i}.
pkgs = []
now = int(datetime.utcnow().timestamp())
now = time.utcnow()
with db.begin():
for i in range(num_packages):
pkgbase = db.create(PackageBase, Name=f"pkg_{i}",
@ -184,7 +183,7 @@ def test_homepage_dashboard(redis, packages, user):
def test_homepage_dashboard_requests(redis, packages, user):
now = int(datetime.utcnow().timestamp())
now = time.utcnow()
pkg = packages[0]
reqtype = db.query(RequestType, RequestType.ID == DELETION_ID).first()
@ -210,7 +209,7 @@ def test_homepage_dashboard_flagged_packages(redis, packages, user):
# Set the first Package flagged by setting its OutOfDateTS column.
pkg = packages[0]
with db.begin():
pkg.PackageBase.OutOfDateTS = int(datetime.utcnow().timestamp())
pkg.PackageBase.OutOfDateTS = time.utcnow()
cookies = {"AURSID": user.login(Request(), "testPassword")}
with client as request:

5
test/test_notify.py
View file

@ -1,11 +1,10 @@
from datetime import datetime
from logging import ERROR
from typing import List
from unittest import mock
import pytest
from aurweb import config, db, models
from aurweb import config, db, models, time
from aurweb.models import Package, PackageBase, PackageRequest, User
from aurweb.models.account_type import TRUSTED_USER_ID, USER_ID
from aurweb.models.request_type import ORPHAN_ID
@ -48,7 +47,7 @@ def user2() -> User:
@pytest.fixture
def pkgbases(user: User) -> List[PackageBase]:
now = int(datetime.utcnow().timestamp())
now = time.utcnow()
output = []
with db.begin():

6
test/test_package_request.py
View file

@ -1,10 +1,8 @@
from datetime import datetime
import pytest
from sqlalchemy.exc import IntegrityError
from aurweb import db
from aurweb import db, time
from aurweb.models.account_type import USER_ID
from aurweb.models.package_base import PackageBase
from aurweb.models.package_request import (ACCEPTED, ACCEPTED_ID, CLOSED, CLOSED_ID, PENDING, PENDING_ID, REJECTED,
@ -54,7 +52,7 @@ def test_package_request_creation(user: User, pkgbase: PackageBase):
def test_package_request_closed(user: User, pkgbase: PackageBase):
ts = int(datetime.utcnow().timestamp())
ts = time.utcnow()
with db.begin():
package_request = db.create(PackageRequest, ReqTypeID=MERGE_ID,
User=user, PackageBase=pkgbase,

6
test/test_package_vote.py
View file

@ -1,10 +1,8 @@
from datetime import datetime
import pytest
from sqlalchemy.exc import IntegrityError
from aurweb import db
from aurweb import db, time
from aurweb.models.account_type import USER_ID
from aurweb.models.package_base import PackageBase
from aurweb.models.package_vote import PackageVote
@ -33,7 +31,7 @@ def pkgbase(user: User) -> PackageBase:
def test_package_vote_creation(user: User, pkgbase: PackageBase):
ts = int(datetime.utcnow().timestamp())
ts = time.utcnow()
with db.begin():
package_vote = db.create(PackageVote, User=user,

25
test/test_packages_routes.py
View file

@ -1,6 +1,5 @@
import re
from datetime import datetime
from http import HTTPStatus
from typing import List
from unittest import mock
@ -9,7 +8,7 @@ import pytest
from fastapi.testclient import TestClient
from aurweb import asgi, db
from aurweb import asgi, db, time
from aurweb.models import License, PackageLicense
from aurweb.models.account_type import USER_ID, AccountType
from aurweb.models.dependency_type import DependencyType
@ -116,7 +115,7 @@ def tu_user():
@pytest.fixture
def package(maintainer: User) -> Package:
""" Yield a Package created by user. """
now = int(datetime.utcnow().timestamp())
now = time.utcnow()
with db.begin():
pkgbase = db.create(PackageBase,
Name="test-package",
@ -138,7 +137,7 @@ def pkgbase(package: Package) -> PackageBase:
@pytest.fixture
def target(maintainer: User) -> PackageBase:
""" Merge target. """
now = int(datetime.utcnow().timestamp())
now = time.utcnow()
with db.begin():
pkgbase = db.create(PackageBase, Name="target-package",
Maintainer=maintainer,
@ -166,7 +165,7 @@ def pkgreq(user: User, pkgbase: PackageBase) -> PackageRequest:
@pytest.fixture
def comment(user: User, package: Package) -> PackageComment:
pkgbase = package.PackageBase
now = int(datetime.utcnow().timestamp())
now = time.utcnow()
with db.begin():
comment = db.create(PackageComment,
User=user,
@ -181,7 +180,7 @@ def comment(user: User, package: Package) -> PackageComment:
def packages(maintainer: User) -> List[Package]:
""" Yield 55 packages named pkg_0 .. pkg_54. """
packages_ = []
now = int(datetime.utcnow().timestamp())
now = time.utcnow()
with db.begin():
for i in range(55):
pkgbase = db.create(PackageBase,
@ -293,7 +292,7 @@ def test_package(client: TestClient, package: Package):
def test_package_comments(client: TestClient, user: User, package: Package):
now = (datetime.utcnow().timestamp())
now = (time.utcnow())
with db.begin():
comment = db.create(PackageComment, PackageBase=package.PackageBase,
User=user, Comments="Test comment", CommentTS=now)
@ -847,7 +846,7 @@ def test_packages_sort_by_popularity(client: TestClient,
def test_packages_sort_by_voted(client: TestClient,
maintainer: User,
packages: List[Package]):
now = int(datetime.utcnow().timestamp())
now = time.utcnow()
with db.begin():
db.create(PackageVote, PackageBase=packages[0].PackageBase,
User=maintainer, VoteTS=now)
@ -942,7 +941,7 @@ def test_packages_sort_by_maintainer(client: TestClient,
def test_packages_sort_by_last_modified(client: TestClient,
packages: List[Package]):
now = int(datetime.utcnow().timestamp())
now = time.utcnow()
# Set the first package's ModifiedTS to be 1000 seconds before now.
package = packages[0]
with db.begin():
@ -970,7 +969,7 @@ def test_packages_flagged(client: TestClient, maintainer: User,
packages: List[Package]):
package = packages[0]
now = int(datetime.utcnow().timestamp())
now = time.utcnow()
with db.begin():
package.PackageBase.OutOfDateTS = now
@ -1100,7 +1099,7 @@ def test_packages_post(client: TestClient, user: User, package: Package):
def test_packages_post_unflag(client: TestClient, user: User,
maintainer: User, package: Package):
# Flag `package` as `user`.
now = int(datetime.utcnow().timestamp())
now = time.utcnow()
with db.begin():
package.PackageBase.Flagger = user
package.PackageBase.OutOfDateTS = now
@ -1127,7 +1126,7 @@ def test_packages_post_unflag(client: TestClient, user: User,
assert successes[0].text.strip() == expected
# Re-flag `package` as `user`.
now = int(datetime.utcnow().timestamp())
now = time.utcnow()
with db.begin():
package.PackageBase.Flagger = user
package.PackageBase.OutOfDateTS = now
@ -1453,7 +1452,7 @@ def test_account_comments(client: TestClient, user: User, package: Package):
""" This test may seem out of place, but it requires packages,
so its being included in the packages routes test suite to
leverage existing fixtures. """
now = (datetime.utcnow().timestamp())
now = time.utcnow()
with db.begin():
# This comment's CommentTS is `now + 1`, so it is found in rendered
# HTML before the rendered_comment, which has a CommentTS of `now`.

6
test/test_packages_util.py
View file

@ -1,10 +1,8 @@
from datetime import datetime
import pytest
from fastapi.testclient import TestClient
from aurweb import asgi, config, db
from aurweb import asgi, config, db, time
from aurweb.models.account_type import USER_ID
from aurweb.models.official_provider import OFFICIAL_BASE, OfficialProvider
from aurweb.models.package import Package
@ -77,7 +75,7 @@ def test_updated_packages(maintainer: User, package: Package):
def test_query_voted(maintainer: User, package: Package):
now = int(datetime.utcnow().timestamp())
now = time.utcnow()
with db.begin():
db.create(PackageVote, User=maintainer, VoteTS=now,
PackageBase=package.PackageBase)

17
test/test_pkgbase_routes.py
View file

@ -1,6 +1,5 @@
import re
from datetime import datetime
from http import HTTPStatus
from typing import List
from unittest import mock
@ -10,7 +9,7 @@ import pytest
from fastapi.testclient import TestClient
from sqlalchemy import and_
from aurweb import asgi, db
from aurweb import asgi, db, time
from aurweb.models.account_type import USER_ID, AccountType
from aurweb.models.dependency_type import DependencyType
from aurweb.models.package import Package
@ -115,7 +114,7 @@ def tu_user():
@pytest.fixture
def package(maintainer: User) -> Package:
""" Yield a Package created by user. """
now = int(datetime.utcnow().timestamp())
now = time.utcnow()
with db.begin():
pkgbase = db.create(PackageBase,
Name="test-package",
@ -137,7 +136,7 @@ def pkgbase(package: Package) -> PackageBase:
@pytest.fixture
def target(maintainer: User) -> PackageBase:
""" Merge target. """
now = int(datetime.utcnow().timestamp())
now = time.utcnow()
with db.begin():
pkgbase = db.create(PackageBase, Name="target-package",
Maintainer=maintainer,
@ -165,7 +164,7 @@ def pkgreq(user: User, pkgbase: PackageBase) -> PackageRequest:
@pytest.fixture
def comment(user: User, package: Package) -> PackageComment:
pkgbase = package.PackageBase
now = int(datetime.utcnow().timestamp())
now = time.utcnow()
with db.begin():
comment = db.create(PackageComment,
User=user,
@ -180,7 +179,7 @@ def comment(user: User, package: Package) -> PackageComment:
def packages(maintainer: User) -> List[Package]:
""" Yield 55 packages named pkg_0 .. pkg_54. """
packages_ = []
now = int(datetime.utcnow().timestamp())
now = time.utcnow()
with db.begin():
for i in range(55):
pkgbase = db.create(PackageBase,
@ -258,7 +257,7 @@ def test_pkgbase_voters(client: TestClient, tu_user: User, package: Package):
pkgbase = package.PackageBase
endpoint = f"/pkgbase/{pkgbase.Name}/voters"
now = int(datetime.utcnow().timestamp())
now = time.utcnow()
with db.begin():
db.create(PackageVote, User=tu_user, PackageBase=pkgbase, VoteTS=now)
@ -279,7 +278,7 @@ def test_pkgbase_voters_unauthorized(client: TestClient, user: User,
pkgbase = package.PackageBase
endpoint = f"/pkgbase/{pkgbase.Name}/voters"
now = int(datetime.utcnow().timestamp())
now = time.utcnow()
with db.begin():
db.create(PackageVote, User=user, PackageBase=pkgbase, VoteTS=now)
@ -303,7 +302,7 @@ def test_pkgbase_comment_not_found(client: TestClient, maintainer: User,
def test_pkgbase_comment_form_unauthorized(client: TestClient, user: User,
maintainer: User, package: Package):
now = int(datetime.utcnow().timestamp())
now = time.utcnow()
with db.begin():
comment = db.create(PackageComment, PackageBase=package.PackageBase,
User=maintainer, Comments="Test",

5
test/test_pkgmaint.py
View file

@ -1,9 +1,8 @@
from datetime import datetime
from typing import List
import pytest
from aurweb import db
from aurweb import db, time
from aurweb.models import Package, PackageBase, User
from aurweb.models.account_type import USER_ID
from aurweb.scripts import pkgmaint
@ -26,7 +25,7 @@ def user() -> User:
def packages(user: User) -> List[Package]:
output = []
now = int(datetime.utcnow().timestamp())
now = time.utcnow()
with db.begin():
for i in range(5):
pkgbase = db.create(PackageBase, Name=f"pkg_{i}",

5
test/test_rendercomment.py
View file

@ -1,9 +1,8 @@
from datetime import datetime
from unittest import mock
import pytest
from aurweb import config, db, logging
from aurweb import config, db, logging, time
from aurweb.models import Package, PackageBase, PackageComment, User
from aurweb.models.account_type import USER_ID
from aurweb.scripts import rendercomment
@ -39,7 +38,7 @@ def user() -> User:
@pytest.fixture
def pkgbase(user: User) -> PackageBase:
now = int(datetime.utcnow().timestamp())
now = time.utcnow()
with db.begin():
pkgbase = db.create(PackageBase, Packager=user, Name="pkgbase_0",
SubmittedTS=now, ModifiedTS=now)

17
test/test_requests.py
View file

@ -1,6 +1,5 @@
import re
from datetime import datetime
from http import HTTPStatus
from logging import DEBUG
from typing import List
@ -10,7 +9,7 @@ import pytest
from fastapi import HTTPException
from fastapi.testclient import TestClient
from aurweb import asgi, config, db, defaults
from aurweb import asgi, config, db, defaults, time
from aurweb.models import Package, PackageBase, PackageRequest, User
from aurweb.models.account_type import TRUSTED_USER_ID, USER_ID
from aurweb.models.package_notification import PackageNotification
@ -94,7 +93,7 @@ def maintainer() -> User:
def packages(maintainer: User) -> List[Package]:
""" Yield 55 packages named pkg_0 .. pkg_54. """
packages_ = []
now = int(datetime.utcnow().timestamp())
now = time.utcnow()
with db.begin():
for i in range(55):
pkgbase = db.create(PackageBase,
@ -148,7 +147,7 @@ def create_pkgbase(user: User, name: str) -> PackageBase:
:param name: PackageBase.Name
:return: PackageBase instance
"""
now = int(datetime.utcnow().timestamp())
now = time.utcnow()
with db.begin():
pkgbase = db.create(PackageBase, Name=name,
Maintainer=user, Packager=user,
@ -185,7 +184,7 @@ def create_request(reqtype_id: int, user: User, pkgbase: PackageBase,
:param comments: PackageRequest.Comments
:return: PackageRequest instance
"""
now = int(datetime.utcnow().timestamp())
now = time.utcnow()
with db.begin():
pkgreq = db.create(PackageRequest, ReqTypeID=reqtype_id,
User=user, PackageBase=pkgbase,
@ -274,7 +273,7 @@ def test_request_post_deletion_autoaccept(client: TestClient, auser: User,
""" Test the request route for deletion as maintainer. """
caplog.set_level(DEBUG)
now = int(datetime.utcnow().timestamp())
now = time.utcnow()
auto_delete_age = config.getint("options", "auto_delete_age")
with db.begin():
pkgbase.ModifiedTS = now - auto_delete_age + 100
@ -503,7 +502,7 @@ def test_orphan_request(client: TestClient, user: User, tu_user: User,
pkgbase: PackageBase, pkgreq: PackageRequest):
""" Test the standard orphan request route. """
idle_time = config.getint("options", "request_idle_time")
now = int(datetime.utcnow().timestamp())
now = time.utcnow()
with db.begin():
pkgreq.ReqTypeID = ORPHAN_ID
# Set the request time so it's seen as due (idle_time has passed).
@ -532,7 +531,7 @@ def test_request_post_orphan_autogenerated_closure(client: TestClient,
pkgbase: PackageBase,
pkgreq: PackageRequest):
idle_time = config.getint("options", "request_idle_time")
now = int(datetime.utcnow().timestamp())
now = time.utcnow()
with db.begin():
pkgreq.ReqTypeID = ORPHAN_ID
# Set the request time so it's seen as due (idle_time has passed).
@ -559,7 +558,7 @@ def test_request_post_orphan_autoaccept(client: TestClient, auser: User,
caplog: pytest.LogCaptureFixture):
""" Test the standard pkgbase request route GET method. """
caplog.set_level(DEBUG)
now = int(datetime.utcnow().timestamp())
now = time.utcnow()
auto_orphan_age = config.getint("options", "auto_orphan_age")
with db.begin():
pkgbase.OutOfDateTS = now - auto_orphan_age - 100

5
test/test_rpc.py
View file

@ -1,6 +1,5 @@
import re
from datetime import datetime
from http import HTTPStatus
from typing import List
from unittest import mock
@ -14,7 +13,7 @@ from redis.client import Pipeline
import aurweb.models.dependency_type as dt
import aurweb.models.relation_type as rt
from aurweb import asgi, config, db, rpc, scripts
from aurweb import asgi, config, db, rpc, scripts, time
from aurweb.models.account_type import USER_ID
from aurweb.models.license import License
from aurweb.models.package import Package
@ -112,7 +111,7 @@ def packages(user: User, user2: User, user3: User) -> List[Package]:
PackageBase=output[0].PackageBase,
Keyword=keyword)
now = int(datetime.utcnow().timestamp())
now = time.utcnow()
for user_ in [user, user2, user3]:
db.create(PackageVote, User=user_,
PackageBase=output[0].PackageBase, VoteTS=now)

5
test/test_rss.py
View file

@ -1,4 +1,3 @@
from datetime import datetime
from http import HTTPStatus
import lxml.etree
@ -6,7 +5,7 @@ import pytest
from fastapi.testclient import TestClient
from aurweb import db, logging
from aurweb import db, logging, time
from aurweb.asgi import app
from aurweb.models.account_type import AccountType
from aurweb.models.package import Package
@ -40,7 +39,7 @@ def user():
@pytest.fixture
def packages(user):
pkgs = []
now = int(datetime.utcnow().timestamp())
now = time.utcnow()
# Create 101 packages; we limit 100 on RSS feeds.
with db.begin():

7
test/test_session.py
View file

@ -1,12 +1,11 @@
""" Test our Session model. """
from datetime import datetime
from unittest import mock
import pytest
from sqlalchemy.exc import IntegrityError
from aurweb import db
from aurweb import db, time
from aurweb.models.account_type import USER_ID
from aurweb.models.session import Session, generate_unique_sid
from aurweb.models.user import User
@ -30,7 +29,7 @@ def user() -> User:
def session(user: User) -> Session:
with db.begin():
session = db.create(Session, User=user, SessionID="testSession",
LastUpdateTS=datetime.utcnow().timestamp())
LastUpdateTS=time.utcnow())
yield session
@ -48,7 +47,7 @@ def test_session_cs():
with db.begin():
session_cs = db.create(Session, User=user2, SessionID="TESTSESSION",
LastUpdateTS=datetime.utcnow().timestamp())
LastUpdateTS=time.utcnow())
assert session_cs.SessionID == "TESTSESSION"
assert session_cs.SessionID != "testSession"

5
test/test_templates.py
View file

@ -1,13 +1,12 @@
import re
from datetime import datetime
from typing import Any, Dict
import pytest
import aurweb.filters # noqa: F401
from aurweb import config, db, templates
from aurweb import config, db, templates, time
from aurweb.filters import as_timezone, number_format
from aurweb.filters import timestamp_to_datetime as to_dt
from aurweb.models import Package, PackageBase, User
@ -59,7 +58,7 @@ def user(db_test) -> User:
@pytest.fixture
def pkgbase(user: User) -> PackageBase:
now = int(datetime.utcnow().timestamp())
now = time.utcnow()
with db.begin():
pkgbase = db.create(PackageBase, Name="test-pkg", Maintainer=user,
SubmittedTS=now, ModifiedTS=now)

15
test/test_trusted_user_routes.py
View file

@ -1,6 +1,5 @@
import re
from datetime import datetime
from http import HTTPStatus
from io import StringIO
from typing import Tuple
@ -10,7 +9,7 @@ import pytest
from fastapi.testclient import TestClient
from aurweb import config, db, filters
from aurweb import config, db, filters, time
from aurweb.models.account_type import DEVELOPER_ID, AccountType
from aurweb.models.tu_vote import TUVote
from aurweb.models.tu_voteinfo import TUVoteInfo
@ -111,7 +110,7 @@ def user():
@pytest.fixture
def proposal(user, tu_user):
ts = int(datetime.utcnow().timestamp())
ts = time.utcnow()
agenda = "Test proposal."
start = ts - 5
end = ts + 1000
@ -165,7 +164,7 @@ def test_tu_empty_index(client, tu_user):
def test_tu_index(client, tu_user):
ts = int(datetime.utcnow().timestamp())
ts = time.utcnow()
# Create some test votes: (Agenda, Start, End).
votes = [
@ -258,7 +257,7 @@ def test_tu_index(client, tu_user):
def test_tu_index_table_paging(client, tu_user):
ts = int(datetime.utcnow().timestamp())
ts = time.utcnow()
with db.begin():
for i in range(25):
@ -366,7 +365,7 @@ def test_tu_index_table_paging(client, tu_user):
def test_tu_index_sorting(client, tu_user):
ts = int(datetime.utcnow().timestamp())
ts = time.utcnow()
with db.begin():
for i in range(2):
@ -436,7 +435,7 @@ def test_tu_index_sorting(client, tu_user):
def test_tu_index_last_votes(client, tu_user, user):
ts = int(datetime.utcnow().timestamp())
ts = time.utcnow()
with db.begin():
# Create a proposal which has ended.
@ -584,7 +583,7 @@ def test_tu_running_proposal(client: TestClient,
def test_tu_ended_proposal(client, proposal):
tu_user, user, voteinfo = proposal
ts = int(datetime.utcnow().timestamp())
ts = time.utcnow()
with db.begin():
voteinfo.End = ts - 5 # 5 seconds ago.

6
test/test_tu_vote.py
View file

@ -1,10 +1,8 @@
from datetime import datetime
import pytest
from sqlalchemy.exc import IntegrityError
from aurweb import db
from aurweb import db, time
from aurweb.models.account_type import TRUSTED_USER_ID
from aurweb.models.tu_vote import TUVote
from aurweb.models.tu_voteinfo import TUVoteInfo
@ -27,7 +25,7 @@ def user() -> User:
@pytest.fixture
def tu_voteinfo(user: User) -> TUVoteInfo:
ts = int(datetime.utcnow().timestamp())
ts = time.utcnow()
with db.begin():
tu_voteinfo = db.create(TUVoteInfo, Agenda="Blah blah.",
User=user.Username,

10
test/test_tu_voteinfo.py
View file

@ -1,10 +1,8 @@
from datetime import datetime
import pytest
from sqlalchemy.exc import IntegrityError
from aurweb import db
from aurweb import db, time
from aurweb.db import create, rollback
from aurweb.models.account_type import TRUSTED_USER_ID
from aurweb.models.tu_voteinfo import TUVoteInfo
@ -26,7 +24,7 @@ def user() -> User:
def test_tu_voteinfo_creation(user: User):
ts = int(datetime.utcnow().timestamp())
ts = time.utcnow()
with db.begin():
tu_voteinfo = create(TUVoteInfo,
Agenda="Blah blah.",
@ -50,7 +48,7 @@ def test_tu_voteinfo_creation(user: User):
def test_tu_voteinfo_is_running(user: User):
ts = int(datetime.utcnow().timestamp())
ts = time.utcnow()
with db.begin():
tu_voteinfo = create(TUVoteInfo,
Agenda="Blah blah.",
@ -66,7 +64,7 @@ def test_tu_voteinfo_is_running(user: User):
def test_tu_voteinfo_total_votes(user: User):
ts = int(datetime.utcnow().timestamp())
ts = time.utcnow()
with db.begin():
tu_voteinfo = create(TUVoteInfo,
Agenda="Blah blah.",

5
test/test_tuvotereminder.py
View file

@ -1,9 +1,8 @@
from datetime import datetime
from typing import Tuple
import pytest
from aurweb import config, db
from aurweb import config, db, time
from aurweb.models import TUVote, TUVoteInfo, User
from aurweb.models.account_type import TRUSTED_USER_ID
from aurweb.scripts import tuvotereminder as reminder
@ -56,7 +55,7 @@ def user3() -> User:
@pytest.fixture
def voteinfo(user: User) -> TUVoteInfo:
now = int(datetime.utcnow().timestamp())
now = time.utcnow()
start = config.getint("tuvotereminder", "range_start")
with db.begin():
voteinfo = db.create(TUVoteInfo, Agenda="Lorem ipsum.",

8
test/test_usermaint.py
View file

@ -1,8 +1,6 @@
from datetime import datetime
import pytest
from aurweb import db
from aurweb import db, time
from aurweb.models import User
from aurweb.models.account_type import USER_ID
from aurweb.scripts import usermaint
@ -24,7 +22,7 @@ def user() -> User:
def test_usermaint_noop(user: User):
""" Last[SSH]Login isn't expired in this test: usermaint is noop. """
now = int(datetime.utcnow().timestamp())
now = time.utcnow()
with db.begin():
user.LastLoginIPAddress = "127.0.0.1"
user.LastLogin = now - 10
@ -45,7 +43,7 @@ def test_usermaint(user: User):
its code path.
"""
now = int(datetime.utcnow().timestamp())
now = time.utcnow()
limit_to = now - 86400 * 7
with db.begin():
user.LastLoginIPAddress = "127.0.0.1"