зеркало из https://github.com/mozilla/bugbot.git
135 строки
4.1 KiB
Python
135 строки
4.1 KiB
Python
# This Source Code Form is subject to the terms of the Mozilla Public
|
|
# License, v. 2.0. If a copy of the MPL was not distributed with this file,
|
|
# You can obtain one at http://mozilla.org/MPL/2.0/.
|
|
|
|
import json
|
|
import unittest
|
|
|
|
import dateutil.parser
|
|
|
|
import bugbot.db as db
|
|
|
|
|
|
class TestDB(unittest.TestCase):
|
|
def by_rule(self, data):
|
|
res = {}
|
|
for x in data:
|
|
rule = x["tool"]
|
|
if rule not in res:
|
|
res[rule] = []
|
|
res[rule].append(x)
|
|
|
|
for rule, info in res.items():
|
|
res[rule] = sorted(info, key=lambda x: dateutil.parser.parse(x["date"]))
|
|
|
|
return res
|
|
|
|
def test_bugchange(self):
|
|
db.session.query(db.BugChange).delete()
|
|
db.session.commit()
|
|
|
|
with open("./tests/data/db_history.json", "r") as In:
|
|
HISTORY = json.load(In)["history"]
|
|
|
|
db.BugChange.import_from_dict(HISTORY)
|
|
|
|
data = db.BugChange.get()
|
|
data = list(data)
|
|
|
|
assert len(data) == len(HISTORY)
|
|
|
|
data = self.by_rule(HISTORY)
|
|
for tool, info in data.items():
|
|
_data = db.BugChange.get(name=tool).order_by(db.BugChange.date.asc())
|
|
_data = list(_data)
|
|
assert len(_data) == len(info)
|
|
|
|
for expected, got in zip(info, _data):
|
|
assert expected["tool"] == got.tool.name
|
|
assert expected["bugid"] == got.bugid
|
|
assert expected["date"] == str(got.get_date())
|
|
|
|
got_extra = got.extra.extra if got.extra else ""
|
|
assert expected["extra"] == got_extra
|
|
|
|
db.BugChange.add("A", 123, ts=123456789, extra="")
|
|
db.BugChange.add("A", 456, ts=123456789, extra="B")
|
|
|
|
data = db.BugChange.get(name="A").order_by(db.BugChange.bugid.asc())
|
|
data = list(data)
|
|
|
|
exp = [
|
|
{"tool": "A", "bugid": 123, "date": 123456789, "extra": ""},
|
|
{"tool": "A", "bugid": 456, "date": 123456789, "extra": "B"},
|
|
]
|
|
|
|
for expected, got in zip(exp, data):
|
|
assert expected["tool"] == got.tool.name
|
|
assert expected["bugid"] == got.bugid
|
|
assert expected["date"] == got.date
|
|
|
|
got_extra = got.extra.extra if got.extra else ""
|
|
assert expected["extra"] == got_extra
|
|
|
|
def test_email(self):
|
|
db.session.query(db.Email).delete()
|
|
db.session.commit()
|
|
|
|
with open("./tests/data/db_history.json", "r") as In:
|
|
EMAILS = json.load(In)["emails"]
|
|
|
|
db.Email.import_from_dict(EMAILS)
|
|
|
|
data = db.Email.get().order_by(db.Email.date.asc())
|
|
data = list(data)
|
|
|
|
assert len(data) == len(EMAILS)
|
|
|
|
for expected, got in zip(EMAILS, data):
|
|
assert expected["tool"] == got.tool.name
|
|
assert expected["user"] == got.user.email
|
|
assert expected["date"] == str(got.get_date())
|
|
|
|
got_res = "Success" if got.result != 0 else "Failure"
|
|
assert expected["result"] == got_res
|
|
|
|
got_extra = got.extra.extra if got.extra else ""
|
|
assert expected["extra"] == got_extra
|
|
|
|
db.Email.add("I", ["J", "K"], "L", "Success", ts=123456789)
|
|
|
|
data = db.Email.get(name="I")
|
|
data = list(data)
|
|
|
|
assert len(data) == 2
|
|
if data[0].user.email == "K":
|
|
data = [data[1], data[0]]
|
|
|
|
exp = [
|
|
{
|
|
"tool": "I",
|
|
"user": "J",
|
|
"date": 123456789,
|
|
"extra": "L",
|
|
"result": "Success",
|
|
},
|
|
{
|
|
"tool": "I",
|
|
"user": "K",
|
|
"date": 123456789,
|
|
"extra": "L",
|
|
"result": "Success",
|
|
},
|
|
]
|
|
|
|
for expected, got in zip(exp, data):
|
|
assert expected["tool"] == got.tool.name
|
|
assert expected["user"] == got.user.email
|
|
assert expected["date"] == got.date
|
|
|
|
got_res = "Success" if got.result != 0 else "Failure"
|
|
assert expected["result"] == got_res
|
|
|
|
got_extra = got.extra.extra if got.extra else ""
|
|
assert expected["extra"] == got_extra
|