GitLab is used only for code review, issue tracking and project management. Canonical locations for source code are still https://gitweb.torproject.org/ https://git.torproject.org/ and git-rw.torproject.org.

...
 
Commits (3)
......@@ -82,5 +82,7 @@ GetTor includes PyTest unit tests. To run the tests, first install the dependenc
```
$ python3 scripts/create_db -n -c -o -f tests/gettor.db
$ python3 scripts/add_links_to_db -f tests/gettor.db
$ pytest-3 tests/
```
......@@ -24,10 +24,8 @@ from email.utils import parseaddr
from twisted.python import log
from twisted.internet import defer
from twisted.enterprise import adbapi
from ..utils.db import SQLite3
from ..utils import strings
from ..utils import validate_email
class AddressError(Exception):
......
......@@ -21,7 +21,6 @@ import configparser
from twisted.python import log
from twisted.internet import defer
from twisted.enterprise import adbapi
from ..utils.db import SQLite3
from ..utils import strings
......
......@@ -12,12 +12,9 @@
from __future__ import absolute_import
import gettext
import hashlib
import configparser
from email import encoders
from email import mime
from email.mime.text import MIMEText
from twisted.internet import defer
......
......@@ -12,7 +12,6 @@
from __future__ import absolute_import
import gettext
import hashlib
import json
import time
......
......@@ -11,9 +11,7 @@
# :license: This is Free Software. See LICENSE for license information.
import os
import time
import logging
import sqlite3
import datetime
import configparser
......
......@@ -11,7 +11,6 @@
# :license: This is Free Software. See LICENSE for license information.
from twisted.logger import Logger
from twisted.enterprise import adbapi
from twisted.application import internet
# Define an application logger
......
......@@ -9,10 +9,10 @@
from __future__ import absolute_import
import sqlite3
from datetime import datetime
from twisted.python import log
from twisted.enterprise import adbapi
class SQLite3(object):
"""
......@@ -20,94 +20,90 @@ class SQLite3(object):
"""
def __init__(self, dbname):
"""Constructor."""
self.dbpool = adbapi.ConnectionPool(
"sqlite3", dbname, check_same_thread=False
)
def query_callback(self, results=None):
"""
Query callback
Log that the database query has been executed and return results
"""
log.msg("Database query executed successfully.")
return results
def query_errback(self, error=None):
"""
Query error callback
Logs database error
"""
if error:
log.msg("Database error: {}".format(error))
return None
self.conn = sqlite3.connect(dbname)
def new_request(self, id, command, service, platform, language, date, status):
"""
Perform a new request to the database
"""
c = self.conn.cursor()
query = "INSERT INTO requests VALUES(?, ?, ?, ?, ?, ?, ?)"
return self.dbpool.runQuery(
query, (id, command, platform, language, service, date, status)
).addCallback(self.query_callback).addErrback(self.query_errback)
c.execute(query, (id, command, platform, language, service,
date, status))
self.conn.commit()
return
def get_requests(self, status, command, service):
"""
Perform a SELECT request to the database
"""
c = self.conn.cursor()
query = "SELECT * FROM requests WHERE service=? AND command=? AND "\
"status = ?"
return self.dbpool.runQuery(
query, (service, command, status)
).addCallback(self.query_callback).addErrback(self.query_errback)
c.execute(query, (service, command, status))
return c.fetchall()
def get_num_requests(self, id, service):
"""
Get number of requests for statistics
"""
query = "SELECT COUNT(rowid) FROM requests WHERE id=? AND service=?"
c = self.conn.cursor()
query = "SELECT COUNT(rowid) FROM requests WHERE id=? AND "\
"service=?"
return self.dbpool.runQuery(
query, (id, service)
).addCallback(self.query_callback).addErrback(self.query_errback)
c.execute(query, (id, service))
return c.fetchone()[0]
def remove_request(self, id, service, date):
"""
Removes completed request record from the database
"""
query = "DELETE FROM requests WHERE id=? AND service=? AND date=?"
c = self.conn.cursor()
query = "DELETE FROM requests WHERE id=? AND service=? AND "\
"date=?"
return self.dbpool.runQuery(
query, (id, service, date)
).addCallback(self.query_callback).addErrback(self.query_errback)
c.execute(query, (id, service, date))
self.conn.commit()
return
def update_stats(self, command, service, platform=None, language='en'):
"""
Update statistics to the database
"""
c = self.conn.cursor()
now_str = datetime.now().strftime("%Y%m%d")
query = "INSERT INTO stats(num_requests, platform, language, command, "\
"service, date) VALUES (1, ?, ?, ?, ?, ?) ON CONFLICT(platform, "\
"language, command, service, date) DO UPDATE SET num_requests=num_requests+1"
"service, date) VALUES (1, ?, ?, ?, ?, ?) ON "\
"CONFLICT(platform, language, command, service, date) "\
"DO UPDATE SET num_requests=num_requests+1"
return self.dbpool.runQuery(
query, (platform, language, command, service, now_str)
).addCallback(self.query_callback).addErrback(self.query_errback)
c.execute(query, (platform, language, command, service,
now_str))
self.conn.commit()
return
def get_links(self, platform, language, status):
"""
Get links from the database per platform
"""
c = self.conn.cursor()
query = "SELECT * FROM links WHERE platform=? AND language=? AND status=?"
return self.dbpool.runQuery(
query, (platform, language, status)
).addCallback(self.query_callback).addErrback(self.query_errback)
c.execute(query, (platform, language, status))
return c.fetchall()
def get_locales(self):
"""
Get a list of the supported tor browser binary locales
"""
c = self.conn.cursor()
query = "SELECT DISTINCT language FROM links"
return self.dbpool.runQuery(query
).addCallback(self.query_callback).addErrback(self.query_errback)
c.execute(query)
locales = []
for locale in c.fetchall():
locales.append(locale[0])
return locales
......@@ -9,8 +9,6 @@ the Tor Browser.
:license: see included LICENSE for information
"""
import argparse
from .settings import Settings
from . import strings
......
......@@ -12,10 +12,6 @@ the Tor Browser.
import json
import os
import platform
import locale
from twisted.python import log
from . import strings
......
......@@ -11,7 +11,6 @@ the Tor Browser.
"""
import json
import locale
import os
import inspect
......
......@@ -20,7 +20,6 @@ import configparser
from time import gmtime, strftime
import core
import utils
"""
GetTor RESTful API
......
......@@ -5,6 +5,7 @@ from __future__ import unicode_literals
from gettor.utils import options
from gettor.utils import strings
from gettor.utils import twitter
from gettor.utils.db import SQLite3
from gettor.services.email.sendmail import Sendmail
from gettor.services.twitter import twitterdm
from gettor.parse.email import EmailParser, AddressError, DKIMError
......
#!/usr/bin/env python3
import pytest
from datetime import datetime
from twisted.trial import unittest
from . import conftests
class DatabaseTests(unittest.TestCase):
# Fail any tests which take longer than 15 seconds.
timeout = 15
def setUp(self):
self.settings = conftests.options.parse_settings("en","tests/test.conf.json")
print(self.settings.get("dbname"))
self.db = conftests.SQLite3(self.settings.get("dbname"))
def tearDown(self):
print("tearDown()")
def add_dummy_requests(self, num):
now_str = datetime.now().strftime("%Y%m%d")
for i in (0, num):
self.db.new_request(
id='testid',
command='links',
platform='linux',
language='en',
service='email',
date=now_str,
status="ONHOLD",
)
def test_stored_locales(self):
locales = self.db.get_locales()
self.assertIn('en-US', locales)
def test_requests(self):
now_str = datetime.now().strftime("%Y%m%d")
self.add_dummy_requests(2)
num = self.db.get_num_requests("testid", "email")
self.assertEqual(num, 2)
requests = self.db.get_requests("ONHOLD", "links", "email")
for request in requests:
print(request)
self.assertEqual(request[1], "links")
self.assertEqual(request[4], "email")
self.assertEqual(request[5], now_str)
self.assertEqual(request[6], "ONHOLD")
self.assertEqual(len(requests), 2)
self.db.remove_request("testid", "email", now_str)
num = self.db.get_num_requests("testid", "email")
self.assertEqual(num, 0)
def test_links(self):
links = self.db.get_links("linux", "en-US", "ACTIVE")
self.assertEqual(len(links), 2) # Right now we have github and gitlab
for link in links:
self.assertEqual(link[1], "linux")
self.assertEqual(link[2], "en-US")
self.assertEqual(link[6], "ACTIVE")
self.assertIn(link[5], ["github", "gitlab"])
if __name__ == "__main__":
unittest.main()