Skip to content
Snippets Groups Projects
Commit ee884833 authored by Thomas Holene Løkkeborg's avatar Thomas Holene Løkkeborg
Browse files

Merge branch 'DatabaseSetup' into 'master'

Database setup

See merge request !10
parents 58608b12 1f300e34
No related branches found
No related tags found
1 merge request!10Database setup
Pipeline #71707 passed
......@@ -35,10 +35,10 @@ tags
# Coc configuration directory
.vim
__pycache__/
# Database
soitool/Database
soitool/database
__pycache__/
# Compiled bytecode of Python source files
*.pyc
......@@ -37,7 +37,7 @@ job_test_gui_ubuntu_vnc:
script:
# -platform because running with a screen is not supported
# https://stackoverflow.com/questions/17106315/failed-to-load-platform-plugin-xcb-while-launching-qt5-app-on-linux-without
- QT_QPA_PLATFORM=vnc python3.7 -m unittest test.test_main
- QT_QPA_PLATFORM=vnc python3.7 -m unittest
job_test_gui_windows:
stage: test
......@@ -45,7 +45,7 @@ job_test_gui_windows:
- ci-windows
script:
- python --version
- python -m unittest test.test_main
- python -m unittest
job_test_gui_ubuntu:
stage: test
......@@ -53,7 +53,7 @@ job_test_gui_ubuntu:
- ci-ubuntu
script:
- python3 --version
- DISPLAY=':10.0' python3 -m unittest test.test_main
- DISPLAY=':10.0' python3 -m unittest
job_pages_smoke_test:
stage: deploy
......
......@@ -410,6 +410,7 @@ good-names=app,
k,
ex,
Run,
f,
_,
x,
y
......
"""Includes all database-related functionality."""
import os
import sqlite3
import json
# Set name and path to (future) database
DBNAME = 'database'
CURDIR = os.path.dirname(__file__)
DBPATH = os.path.join(CURDIR, DBNAME)
# DDL-statements for creating tables
CODEBOOK = 'CREATE TABLE CodeBook' \
'(Word VARCHAR PRIMARY KEY, Category VARCHAR,' \
' Type int DEFAULT 0, Code VARCHAR)'
CATEGORYWORDS = 'CREATE TABLE CategoryWords' \
'(Word VARCHAR PRIMARY KEY, Category VARCHAR)'
BYHEART = 'CREATE TABLE ByHeart(Word VARCHAR PRIMARY KEY)'
class Database():
"""
Holds database-connection and related functions.
Connects to existing db if found, creates new db if not.
If db is created, tables are created and filled.
"""
def __init__(self):
db_exists = os.path.exists(DBPATH)
if db_exists:
print('Connecting to existing DB.')
self.conn = sqlite3.connect(DBPATH)
print('DB-connection established.')
else:
print('Creating new DB.')
self.conn = sqlite3.connect(DBPATH)
self.create_tables()
print('DB created.')
self.fill_tables()
print('Tables filled with data.')
self.conn.row_factory = sqlite3.Row # Enables row['columnName']
def create_tables(self):
"""Create tables CodeBook, CategoryWords and ByHeart."""
stmts = [CODEBOOK, CATEGORYWORDS, BYHEART]
for stmt in stmts:
self.conn.execute(stmt)
self.conn.commit()
def fill_tables(self):
"""Fill tables with testdata."""
self.fill_codebook()
self.fill_by_heart()
self.fill_category_words()
self.conn.commit()
def fill_codebook(self):
"""Read data from codebook.json and fills DB-table CodeBook."""
file_path = os.path.join(CURDIR, "testdata/codebook.json")
# Load json as dict
with open(file_path, "r") as file:
codewords = json.load(file)
file.close()
# Insert data in db
stmt = 'INSERT INTO CodeBook(Word, Category, Type, Code)' \
'VALUES(?,?,?,?)'
for word in codewords:
self.conn.execute(stmt, (word['word'], word['category'],
word['type'], word['code']))
def fill_by_heart(self):
"""Read data from ByHeart.txt and fill DB-table ByHeart."""
file_path = os.path.join(CURDIR, "testdata/ByHeart.txt")
f = open(file_path, "r", encoding='utf-8')
# Loop through words on file and insert them into ByHeart-table
stmt = 'INSERT INTO ByHeart(Word) VALUES(?)'
for expr in f:
self.conn.execute(stmt, (expr.rstrip(),))
f.close()
def fill_category_words(self):
"""Read data from CategoryWords.txt and fill DB-table CategoryWords."""
file_path = os.path.join(CURDIR, "testdata/CategoryWords.txt")
f = open(file_path, "r", encoding='utf-8')
# Get number of categories on file
no_of_categories = int(f.readline().rstrip())
# Loop through categories on file
for _ in range(no_of_categories):
# Get category and number of words in category
line = f.readline().split(", ")
category = line[0]
no_of_words = int(line[1].rstrip())
# Loop through words in category and add rows to DB
stmt = 'INSERT INTO CategoryWords(Word, Category) VALUES(?, ?)'
for _ in range(no_of_words):
word = f.readline().rstrip()
self.conn.execute(stmt, (word, category,))
f.close()
def get_categories(self):
"""
Retrieve all categories from DB-table CategoryWords.
Returns
-------
List of strings
Categories
"""
stmt = 'SELECT DISTINCT Category FROM CategoryWords'
queried = self.conn.execute(stmt)
categories = []
for row in queried:
categories.append(row['Category'])
return categories
def get_codebook(self, small=False):
"""
Retrieve the entries belonging to the full or small codebook.
Parameters
----------
small : Bool
Full or small codebook to be returned
Returns
-------
codebook : list (of dicts)
[{'word': str, 'type': str, 'category': str, 'code': str}]
"""
# Get either full or small codebook
stmt = 'SELECT * FROM CodeBook'
if small:
stmt = stmt + ' WHERE Type = ?'
queried = self.conn.execute(stmt, (1,)).fetchall()
else:
queried = self.conn.execute(stmt).fetchall()
codebook = []
for entry in queried:
codebook.append({'word': entry['Word'],
'category': entry['Category'],
'type': entry['Type'],
'code': entry['Code']})
return codebook
START RADIOTAUSHET
SLUTT RADIOTAUSHET
SOI KOMPROMITTERT
6
Hunderase, 8
Chihuahua
Terrier
Dalmantiner
Husky
Puddel
Schæfer
Collie
Labrador
Bilmerke, 10
Volkswagen
Audi
Toyota
Ferrari
Suzuki
Honda
Ford
Opel
Mazda
Subaru
Land, 16
Norge
Sverige
Frankrike
Italia
Russland
Kina
Belgia
Canada
Australia
Argentina
India
Danmark
Tyrkia
Spania
Thailand
Tyskland
Tresort, 7
Eik
Furu
Gran
Ask
Bøk
Lønn
Bjørk
President, 7
Obama
Trump
Franklin
Bush
Clinton
Nixon
Kennedy
Some, 5
Facebook
TikTok
Tinder
Snapchat
Instagram
[
{"word":"Vann (l)","category":"Etterforsyninger", "type":0, "code": "1A"},
{"word":"Meldingsblankett (blokker)", "category":"Etterforsyninger", "type":0, "code":"1B"},
{"word":"PACE-batteri/BA-3090 (stk)", "category":"Etterforsyninger", "type":0, "code":"1C"},
{"word":"Rødsprit (l)", "category":"Etterforsyninger", "type":0, "code":"1D"},
{"word":"Proviant (DOS)", "category":"Etterforsyninger", "type":0, "code":"1E"},
{"word":"Kryss", "category":"Landemerker", "type":0, "code":"2A"},
{"word":"Sti", "category":"Landemerker", "type":0, "code":"2B"},
{"word":"Veg", "category":"Landemerker", "type":0, "code":"2C"},
{"word":"Høyde", "category":"Landemerker", "type":0, "code":"2D"},
{"word":"Rute", "category":"Landemerker", "type":0, "code":"2E"},
{"word":"Elv", "category":"Landemerker", "type":0, "code":"2F"},
{"word":"Dal", "category":"Landemerker", "type":0, "code":"2G"},
{"word":"Bro", "category":"Landemerker", "type":0, "code":"2H"},
{"word":"Lysrakett", "category":"Våpenteknisk", "type":0, "code":"3A"},
{"word":"40 mm", "category":"Våpenteknisk", "type":0, "code":"3B"},
{"word":"Håndgranat", "category":"Våpenteknisk", "type":0, "code":"3C"},
{"word":"P-80", "category":"Våpenteknisk", "type":0, "code": "3D"},
{"word":"Bombe", "category":"Våpenteknisk", "type":0, "code":"3E"},
{"word":"Ammo", "category":"Våpenteknisk", "type":0, "code":"3F"},
{"word":"Signalpistol", "category":"Våpenteknisk", "type":0, "code":"3G"},
{"word":"ERYX", "category":"Våpenteknisk", "type":0, "code":"3H"},
{"word":"MP-5", "category":"Våpenteknisk", "type":1, "code":"3I"},
{"word":"HK-416", "category":"Våpenteknisk", "type":0, "code":"3J"},
{"word":"Går av nett", "category":"Uttrykk/tiltak/oppdrag", "type":0, "code":"4A"},
{"word":"Er i rute", "category":"Uttrykk/tiltak/oppdrag", "type":0, "code":"4B"},
{"word":"Stans, -e, -et", "category":"Uttrykk/tiltak/oppdrag", "type":0, "code":"4C"}
]
\ No newline at end of file
"""Database tests."""
import os
from pathlib import Path
import unittest
import json
from soitool.database import Database
TESTDATA_PATH = Path(__file__).parent.parent / "soitool/testdata"
class DatabaseTest(unittest.TestCase):
"""Database tests."""
def setUp(self):
"""Connect to/create database."""
self.database = Database()
def test_connection(self):
"""Assert connection is not None."""
self.assertIsNotNone(self.database)
def test_by_heart(self):
"""Assert contents of table ByHeart in DB matches testdata."""
# Open and read file:
file_path = os.path.join(TESTDATA_PATH, "ByHeart.txt")
f = open(file_path, "r", encoding="utf-8")
file_content = f.read()
# Retrieve expressions from file:
expressions = file_content.split("\n")[:-1] # ignore bottom line
no_of_expr = len(expressions)
# Retrieve expressions from DB:
stmt = 'SELECT * FROM ByHeart'
queried = self.database.conn.execute(stmt).fetchall()
# Assert equal amount of expressions in table and file
no_of_expr_db = len(queried)
self.assertEqual(no_of_expr_db, no_of_expr)
# Assert equal expressions in table and file
for i, expr in enumerate(expressions):
self.assertEqual(queried[i][0], expr)
f.close()
def test_category_words(self):
"""Assert contents of table CategoryWords in DB matches testdata."""
file_path = os.path.join(TESTDATA_PATH, "CategoryWords.txt")
f = open(file_path, "r", encoding="utf-8")
categories_file = []
words_file = []
# Get number of categories on file
no_of_categories = int(f.readline().rstrip("\\n"))
# Loop through categories on file
for _ in range(no_of_categories):
# Get category and number of words in category
line = f.readline().split(",")
categories_file.append(line[0])
no_of_words = int(line[1][:-1])
# Loop through words in category
for _ in range(no_of_words):
words_file.append(f.readline().rstrip())
f.close()
# Assert equal categories in table and file
categories_db = self.database.get_categories()
self.assertEqual(categories_db, categories_file)
# Retrieve data from DB
stmt = 'SELECT * FROM CategoryWords'
queried = self.database.conn.execute(stmt).fetchall()
# Assert equal categories in table and file
words_db = [row[0] for row in queried]
self.assertEqual(words_db, words_file)
def test_codebook(self):
"""Assert function get_codebook works as expected."""
# Get test-data from json
file_path = os.path.join(TESTDATA_PATH, "codebook.json")
with open(file_path, "r") as file:
expected = json.load(file)
file.close()
# Get data from db
stmt = 'SELECT * FROM CodeBook'
actual = self.database.conn.execute(stmt).fetchall()
# Check same lenght
self.assertEqual(len(expected), len(actual))
# Check total equality
for i, entry in enumerate(expected):
self.assertEqual(entry['word'], actual[i][0])
self.assertEqual(entry['category'], actual[i][1])
self.assertEqual(entry['type'], actual[i][2])
self.assertEqual(entry['code'], actual[i][3])
def test_get_categories(self):
"""Assert function get_categories works as expected."""
file_path = os.path.join(TESTDATA_PATH, "CategoryWords.txt")
f = open(file_path, "r", encoding="utf-8")
# Get number of categories on file
no_of_categories = int(f.readline().rstrip())
categories_file = []
for _ in range(no_of_categories):
line = f.readline().split(", ")
categories_file.append(line[0])
# Skip all words:
for _ in range(int(line[1].rstrip())):
f.readline()
f.close()
# Assert categories are equal
categories_db = self.database.get_categories()
self.assertEqual(categories_file, categories_db)
def test_get_codebook(self):
"""Assert function get_codebook returns full codebook."""
# Load full codebook
file_path = os.path.join(TESTDATA_PATH, "codebook.json")
with open(file_path, "r") as file:
expected = json.load(file)
file.close()
# Get full codebook from db
actual = self.database.get_codebook()
# Compare lenth
self.assertEqual(len(expected), len(actual))
# Compare contents
for i, entry in enumerate(expected):
self.assertEqual(entry['word'], actual[i]['word'])
self.assertEqual(entry['category'], actual[i]['category'])
self.assertEqual(entry['type'], actual[i]['type'])
self.assertEqual(entry['code'], actual[i]['code'])
def test_get_codebook_small(self):
"""Assert function get_codebook only return the small codebook."""
# Load full codebook
file_path = os.path.join(TESTDATA_PATH, "codebook.json")
with open(file_path, "r") as file:
data = json.load(file)
file.close()
# Fill expected with only small codebook entries
expected = []
for entry in data:
if entry['type'] == 1:
expected.append(entry)
# Get small codebook from db
actual = self.database.get_codebook(small=True)
# Compare lenght
self.assertEqual(len(expected), len(actual))
# Compare contents
for i, entry in enumerate(expected):
self.assertEqual(entry['word'], actual[i]['word'])
self.assertEqual(entry['category'], actual[i]['category'])
self.assertEqual(entry['type'], actual[i]['type'])
self.assertEqual(entry['code'], actual[i]['code'])
if __name__ == '__main__':
unittest.main()
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment