Testing changes to a graph

test_graph_changes.py

import rdflib as R
from random import randint,sample
import subprocess
from subprocess import PIPE
import os
# Load in the graph
g = R.ConjunctiveGraph(identifier=R.URIRef("http://somehost.com/default_graph"))
graph_uri = R.URIRef("http://somehost.com/default_graph")
YOUR_TEST_GRAPH = 'changes_test.nq'
YOUR_TEST_GRAPH_FORMAT = 'nquads'
YOUR_TEST_OUPUT_FORMAT = 'trig'
g.parse(YOUR_TEST_GRAPH,format=YOUR_TEST_GRAPH_FORMAT)
def trip(lower,upper):
    # Determines the range of values that may be taken.
    # (upper - lower) indicates how likely it is that triples share some resource
    r = randint(lower,upper)
    s = R.URIRef("http://somehost.com/s" + str(r))
    r = randint(lower,upper)
    p = R.URIRef("http://somehost.com/p" + str(r))
    r = randint(lower,upper)
    o = R.URIRef("http://somehost.com/o" + str(r))
    return (s,p,o,graph_uri)

def remove_random(n):
    to_remove = sample(g,n)
    for x in to_remove:
        g.remove(x)

try:
    os.unlink('jack')
    os.unlink('jill')
except:
    pass

open('jill','w').close()
diff_file = open('diffs','w')

for x in range(10):
    add_num = 200
    rem_num = 200
    for _ in range(add_num):
        t = trip(0,add_num / 3)
        g.add(t)
    remove_random(rem_num)
    f = open('temp','w')
    g.serialize(f,format=YOUR_TEST_OUPUT_FORMAT)
    f.close()

    f = open('jack','w')
    subprocess.call('sort temp', shell=True, stdout=f)
    f.close()

    with subprocess.Popen('diff jill jack', shell=True,stdout=PIPE).stdout as diff_out:
        diff_file.write(diff_out.read())
    diff_file.write("=========================\n")
    os.rename('jack', 'jill')
diff_file.close()

# several times:
#    Add some triples
#    Remove some triples
#    write this graph's serialization
#    Print out the diff between this and the last iteration
#    replace the old serialization with this one