1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
|
import unittest
import sys, logging
import freebase
import random
import time
import getlogindetails
from freebase.api import HTTPMetawebSession, MetawebError
from freebase.schema import dump_type, dump_base, restore
USERNAME = 'username'
PASSWORD = 'password'
API_HOST = 'sandbox-freebase.com'
s = freebase.api.HTTPMetawebSession(API_HOST)
if USERNAME == "username" and PASSWORD == "password":
USERNAME, PASSWORD = getlogindetails.main()
s.login(USERNAME, PASSWORD)
# Sorry, this is just so annoying to type.
f = lambda x: x["id"]
class TestHardcoreSchemaManipulation(unittest.TestCase):
def test_copy_an_entire_domain(self):
domain_id = _create_domain()
ex_domain_id = "/base/contractbridge" # example domain id
ex_domain_type = "bridge_player"
ex_domain_type_id = ex_domain_id + "/" + ex_domain_type
graph = dump_base(s, ex_domain_id)
restore(s, graph, domain_id)
newperson, realperson = s.mqlreadmulti([{"id" : domain_id + "/" + ex_domain_type, "/type/type/properties" : {"return" : "count" }},
{"id" : ex_domain_type_id, "/type/type/properties" : {"return" : "count" }}])
self.assertEqual(newperson["/type/type/properties"], realperson["/type/type/properties"])
# let's try and check everything.
# - check all the types are there
realtypes = s.mqlread([{"id" : None, "type" : "/type/type", "domain" : ex_domain_id}])
newtypes = s.mqlread([{"id" : None, "type" : "/type/type", "domain" : domain_id}])
l = lambda q: sorted(map(f, q))
realtypes = l(realtypes)
newtypes = l(newtypes)
self.assertEqual(len(realtypes), len(newtypes))
for i in range(len(realtypes)):
self.assertEqual(realtypes[i].rsplit("/", 1)[-1], newtypes[i].rsplit("/", 1)[-1])
# - check the properties are the same
def get_properties(types):
properties = set()
for i in s.mqlreadmulti([{"id" : id, "/type/type/properties" : [{"id" : None}]} for id in types]):
properties.update(map(lambda x: x["id"], i["/type/type/properties"]))
return properties
realproperties = sorted(list(get_properties(realtypes)))
newproperties = sorted(list(get_properties(newtypes)))
def ignore_base(id): return id.rsplit("/", 1)[-1]
self.assertEqual(len(realproperties), len(newproperties))
self.assertEqual([ignore_base(prop_id) for prop_id in realproperties], [ignore_base(prop_id) for prop_id in newproperties])
# - check the properties and type's attributes are the same
# TODO
def test_restore_over_restore(self):
domain_id = _create_domain()
graph = dump_domain(s, "/base/contractbridge")
restore(s, graph, domain_id)
# now we restore again... it should raise a MetawebError
self.assertRaises(MetawebError, restore(s, graph, domain_id))
def test_try_copying_a_cvt(self):
# if follow_types is True, everything is kosher.
domain_id = _create_domain()
graph = dump_type(s, "/film/actor", follow_types=True)
restore(s, graph, domain_id)
newactor, realactor = s.mqlreadmulti([{"id" : domain_id + "/actor", "/type/type/properties" : {"return" : "count" }},
{"id" : "/film/actor", "/type/type/properties" : {"return" : "count" }}])
self.assertEqual(newactor["/type/type/properties"], realactor["/type/type/properties"])
# if follow_types is False, if we try to upload a cvt, it should whine
from freebase.schema import CVTError
self.assertRaises(CVTError, lambda: dump_type(s, "/film/actor", follow_types=False))
def _create_domain():
domain_id = s.create_private_domain("test" + str(int(random.random() * 1e10)), "test")["domain_id"]
domain_id = s.mqlread({"id" : domain_id, "a:id" : None})["a:id"]
return domain_id
if __name__ == '__main__':
if USERNAME == "username" and PASSWORD == "password":
try:
passwordfile = open(".password.txt", "r")
fh = passwordfile.read().split("\n")
USERNAME = fh[0]
PASSWORD = fh[1]
passwordfile.close()
s.login(USERNAME, PASSWORD)
except Exception, e:
print "In order to run the tests, we need to use a valid freebase username and password"
USERNAME = raw_input("Please enter your username: ")
PASSWORD = raw_input("Please enter your password (it'll appear in cleartext): ")
s.login(USERNAME, PASSWORD)
print "Thanks!"
else:
s.login(USERNAME, PASSWORD)
unittest.main()
|
