datajoint-python/tests/test_update1.py at master · mspacek/datajoint-python · GitHub

Repository navigation

History

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

from nose.tools import assert_true, assert_false, assert_equal, raises

import os

import numpy as np

from pathlib import Path

import tempfile

import datajoint as dj

from . import PREFIX, CONN_INFO

from datajoint import DataJointError

schema = dj.Schema(PREFIX + '_update1', connection=dj.conn(**CONN_INFO))

dj.config['stores']['update_store'] = dict(

protocol='file',

location=tempfile.mkdtemp())

dj.config['stores']['update_repo'] = dict(

stage=tempfile.mkdtemp(),

protocol='file',

location=tempfile.mkdtemp())

scratch_folder = tempfile.mkdtemp()

dj.errors._switch_filepath_types(True)

@schema

class Thing(dj.Manual):

definition = """

thing : int

---

number=0 : int

frac : float

picture = null : attach@update_store

params = null : longblob

img_file = null: filepath@update_repo

timestamp = CURRENT_TIMESTAMP : datetime

"""

def test_update1():

"""test normal updates"""

dj.errors._switch_filepath_types(True)

# CHECK 1 -- initial insert

key = dict(thing=1)

Thing.insert1(dict(key, frac=0.5))

check1 = Thing.fetch1()

# CHECK 2 -- some updates

# numbers and datetimes

Thing.update1(dict(key, number=3, frac=30, timestamp="2020-01-01 10:00:00"))

# attachment

attach_file = Path(scratch_folder, 'attach1.dat')

buffer1 = os.urandom(100)

attach_file.write_bytes(buffer1)

Thing.update1(dict(key, picture=attach_file))

attach_file.unlink()

assert_false(attach_file.is_file())

# filepath

stage_path = dj.config['stores']['update_repo']['stage']

relpath, filename = 'one/two/three', 'picture.dat'

managed_file = Path(stage_path, relpath, filename)

managed_file.parent.mkdir(parents=True, exist_ok=True)

original_file_data = os.urandom(3000)

with managed_file.open('wb') as f:

f.write(original_file_data)

Thing.update1(dict(key, img_file=managed_file))

managed_file.unlink()

assert_false(managed_file.is_file())

check2 = Thing.fetch1(download_path=scratch_folder)

buffer2 = Path(check2['picture']).read_bytes() # read attachment

final_file_data = managed_file.read_bytes() # read filepath

# CHECK 3 -- reset to default values using None

Thing.update1(dict(key, number=None, timestamp=None, picture=None, img_file=None, params=np.random.randn(3, 3)))

check3 = Thing.fetch1()

assert_true(check1['number'] == 0 and

check1['picture'] is None and

check1['params'] is None)

assert_true(check2['number'] == 3 and

check2['frac'] == 30.0 and

check2['picture'] is not None and

check2['params'] is None and buffer1==buffer2)

assert_true(check3['number'] == 0 and

check3['frac'] == 30.0 and

check3['picture'] is None and

check3['img_file'] is None and

isinstance(check3['params'], np.ndarray))

assert_true(check3['timestamp'] > check2['timestamp'])

assert_equal(buffer1, buffer2)

assert_equal(original_file_data, final_file_data)

@raises(DataJointError)

def test_update1_nonexistent():

Thing.update1(dict(thing=100, frac=0.5)) # updating a non-existent entry

@raises(DataJointError)

def test_update1_noprimary():

Thing.update1(dict(number=None)) # missing primary key

@raises(DataJointError)

def test_update1_misspelled_attribute():

key = dict(thing=17)

Thing.insert1(dict(key, frac=1.5))

Thing.update1(dict(key, numer=3)) # misspelled attribute

datajoint-python/tests/test_update1.py at master · mspacek/datajoint-python · GitHub

Repository navigation

Expand file tree

Breadcrumbs

test_update1.py

Latest commit

History

Breadcrumbs

test_update1.py

File metadata and controls