Test Upsert Operations

Setup

from fastcore.parallel import parallel
from fastcore.test import *
from fastsql import *
from sqlalchemy.exc import CompileError, ProgrammingError

Note: Make sure to use fastsql’s database() here

db = database(':memory:')

class People: id: int; name: str

people = db.create(People, pk='id')

Test Single Upserts

Here we test upsert()

Test Cases for upsert() Where Nothing Is Inserted

Test that calling upsert() without any parameters doesn’t change anything, and returns nothing

people.upsert()

{}

Test None doesn’t change anything.

count = people.count
assert people.upsert(None) == {}
assert people.count == count

Test empty dict doesn’t change anything

count = people.count
assert people.upsert({}) == {}
assert people.count == count

# Test empty dataclass doesn't change anything
PersonDC = people.dataclass()
count = people.count
assert people.upsert(PersonDC()) == {}
assert people.count == count

# Test empty class instance doesn't change anything
class EmptyPerson: pass
count = people.count
assert people.upsert(EmptyPerson()) == {}
assert people.count == count

Single Insert Types

Test upsert with keyword argument without id. Result should be a MissingPrimaryKey error

try: people.upsert(name='Alice')
except (MissingPrimaryKey,KeyError): print('Correct throwing of key error')

Correct throwing of key error

Use upsert to insert a new record via a dataclass. Since it can’t find the id, it adds the record

person = people.upsert(People(name='Alice', id=people.count+1))

Test upsert that updates with dataclass. Since it can find the id, it updates the record.

assert people.upsert(People(name='Bobba', id=person.id)).name == 'Bobba'

Use upsert to insert a new record via a class. Since it can’t find the id, it adds the record

count = people.count
class Student: pass
student = Student()
student.name = 'Daniel Greenfeld'
student.id = people.count+1

assert people.upsert(student).name == 'Daniel Greenfeld'
assert people.count == count+1

Test upsert that updates with class. Since it can find the id, it updates the record.

count = people.count
student = Student()
student.name = 'Daniel Roy Greenfeld'
student.id = person.id

assert people.upsert(student).name == 'Daniel Roy Greenfeld'
assert people.count == count

None and Empty String Handling

Test upserting a record with name set to None. First assert checks the method result, the second assert tests that the database was altered correctly.

result = people.upsert(People(name=None, id=person.id))
assert result.name is None
assert people[person.id].name is None

Test with empty string.

result = people.upsert(People(name='', id=person.id))
assert result.name == ''
assert people[person.id].name == ''

Other Cases

Test upserts with special characters. Let’s do updates first

assert people.upsert(People(name='O\'Connor', id=person.id)).name == "O'Connor"
assert people[person.id].name == "O'Connor"
assert people.upsert(People(name='José', id=person.id)).name == 'José'
assert people[person.id].name == "José"

Now test special characters with upserts that insert.

person = people.upsert(People(name='O\'Connor', id=people.count+1))
assert person.name == "O'Connor"
assert people[person.id].name == "O'Connor"
person = people.upsert(People(name='José', id=people.count+1))
assert person.name == "José"
assert people[person.id].name == "José"

Test dict upsert

assert people.upsert({'name': 'Dict Test', 'id': person.id}).name == 'Dict Test'

test_fail(people.upsert, kwargs=dict(name='Extra', age=25, title='Dr', id=person.id), exc=CompileError)

Test that threaded concurrency fails on writes

def up(i): return people.upsert(People(id=i, name=f'Person {i}'))
test_fail(parallel, args=[up, range(1, 100)], kwargs={'threadpool': True}, exc=ProgrammingError)