def __init__(self):
- # a dictonary holding the various attribute indexes. The keys
- # are lowercase attribute names, values are MemIndex or
- # CidrMemIndex objects.
+ # a dictionary holding the various attribute indexes. The keys
+ # are lowercase attribute names, values are MemIndex,
+ # CidrMemIndex, or ComboMemIndex objects.
self.indexes = {}
- # a dictonary holding the actual rwhoisobjects. keys are
+ # a dictionary holding the actual rwhoisobjects. keys are
# string IDs, values are rwhoisobject instances.
self.main_index = {}
self.normal_indexes = []
self.cidr_indexes = []
- # dictonary holding all of the seen class names. keys are
+ # dictionary holding all of the seen class names. keys are
# lowercase classnames, value is always None.
self.classes = {}
def init_schema(self, schema_file):
"""Initialize the schema from a schema file. Currently the
schema file is a list of 'attribute_name = index_type' pairs,
- one per line. index_type is one of N or C, where N means a
- normal string index, and C means a CIDR index.
+ one per line. index_type is one of A (all), N (normal) or C
+ (cidr).
It should be noted that this database implementation
implements a global namespace for attributes, which isn't
# initialize base schema
- self.attrs['id'] = "N"
- self.attrs['auth-area'] = None
+ self.attrs['id'] = "N"
+ self.attrs['auth-area'] = None
self.attrs['class-name'] = None
- self.attrs['updated'] = None
+ self.attrs['updated'] = None
self.attrs['referred-auth-area'] = "R"
sf = open(schema_file, "r")
line = line.strip()
if not line or line.startswith("#"): continue
- attr, it = line.split("=")
+ attr, it = line.split("=", 1)
self.attrs[attr.strip().lower()] = it.strip()[0].upper()
for attr, index_type in self.attrs.items():
self.cidr_indexes.append(attr)
elif index_type == "R":
# referral index, an all index that must be searched
- # explictly by attribute
+ # explicitly by attribute
self.indexes[attr] = MemIndex.ComboMemIndex()
elif index_type == "C":
# a cidr index
def is_indexed_attr(self, attr):
if self.is_attribute(attr):
- return self.attrs[attr.lower()]
+ return self.indexes.has_key(attr.lower())
return False
def is_objectclass(self, objectclass):
return res
def search_referral(self, value, max = 0):
- """Given a heirarchal value, search for referrals. Returns a
+ """Given a hierarchical value, search for referrals. Returns a
list of object ids or an empty list."""
return self.search_attr("referred-auth-area", value, max)
to_del = self.data[n:]
for i in to_del: del self._dict[i]
self.data = self.data[:n]
-
-
-# test driver
-if __name__ == "__main__":
- import sys
- db = MemDB()
-
- print "loading schema:", sys.argv[1]
- db.init_schema(sys.argv[1])
- for data_file in sys.argv[2:]:
- print "loading data file:", data_file
- db.load_data(data_file)
- db.index_data()
-
- print "Schema: authority areas"
- for a in db.authareas.keys():
- print " %s" % a
- print "Schema: classes"
- for c in db.classes.keys():
- print " %s" % c
- print "Schema: attributes"
- for a in db.attrs.keys():
- print " %s" % a
-
- print "Is 'Network' a class?", db.is_objectclass("Network")
-
-# for k, v in db.main_index.items():
-# print "main_index[", k, "]:", v
-
- print "searching for a.com"
- res = db.search_attr("domain-name", "a.com")
- print res.list()
- print [ str(x) for x in db.fetch_objects(res.list()) ]
-
- print "searching for doe"
- res = db.search_normal("doe")
- print res.list()
- print [ str(x) for x in db.fetch_objects(res.list()) ]
-
- print "searching for 10.0.0.2"
- res = db.search_cidr("10.0.0.2")
- print res.list()
- print [ str(x) for x in db.fetch_objects(res.list()) ]
-
- print "searching for fddi.a.com"
- res = db.search_normal("fddi.a.com")
- print res.list()
-
- print "searching referral index for fddi.a.com"
- res = db.search_attr("referred-auth-area", "fddi.a.com")
- print res.list()
- print [ str(x) for x in db.fetch_objects(res.list()) ]
-
-
--- /dev/null
+# This file is part of python-rwhoisd
+#
+# Copyright (C) 2008 David E. Blacka
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation; either version 2 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful, but
+# WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+# General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program; if not, write to the Free Software
+# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307
+# USA
+
+
+import path
+import MemDB, Rwhois
+import unittest, types
+
+class InitSchemaTest(unittest.TestCase):
+
+ def testInitSchema(self):
+ db = MemDB.MemDB()
+ assert(db)
+
+ db.init_schema("test_schema")
+
+ assert(db.is_attribute("id")) # built-in
+ assert(db.is_attribute("auth-area")) # built-in
+ assert(db.is_attribute("class-name")) # built-in
+ assert(db.is_attribute("updated")) # built-in
+ assert(db.is_attribute("referred-auth-area")) # built-in
+ assert(db.is_attribute("name"))
+ assert(db.is_attribute("ip-network"))
+ assert(db.is_indexed_attr("id"))
+ assert(db.is_indexed_attr("ip-network"))
+ assert(db.is_indexed_attr("ref-like"))
+
+ # At the moment, init_schema doesn't really fail on malformed
+ # schema files. Instead, it just silently fails to add an index
+ # for the mailformed attributes.
+ def testBadSchema(self):
+ db = MemDB.MemDB()
+ assert(db)
+
+ db.init_schema("test_bad_schema")
+
+ assert(db.is_attribute("name"))
+ assert(not db.is_indexed_attr("name"))
+ assert(db.is_attribute("bad"))
+ assert(not db.is_indexed_attr("bad"))
+ assert(db.is_indexed_attr("questionable"))
+
+
+class SchemaTest(unittest.TestCase):
+
+ def setUp(self):
+ self.db = MemDB.MemDB()
+ assert(self.db)
+ self.db.init_schema("test_schema")
+ self.db.load_data("test_data")
+ self.db.index_data()
+
+ auth_areas = ["10.0.0.0/8", "a.com"]
+
+ def testAuthAreas(self):
+ a = set()
+ b = set()
+ for aa in self.auth_areas:
+ a.add(aa)
+ assert(self.db.is_autharea(aa))
+ for aa in self.db.get_authareas():
+ b.add(aa)
+
+ self.assertEquals(a, b)
+
+ classes = ['contact', 'domain', 'host', 'network',
+ 'organization', 'referral']
+
+ def testClasses(self):
+ for c in self.classes:
+ assert(self.db.is_objectclass(c))
+
+ attrs = ['Admin-Contact', 'Auth-Area', 'Billing-Contact', 'City',
+ 'Class-Name', 'Country-Code', 'Created', 'Domain-Name',
+ 'Email', 'Fax', 'First-Name', 'Guardian', 'Host-Name',
+ 'ID', 'IP-Address', 'IP-Network', 'Last-Name', 'Name',
+ 'Network-Block', 'Network-Name', 'Organization', 'Org-Name',
+ 'Phone', 'Postal-Code', 'Primary-Server', 'Referral',
+ 'Referred-Auth-Area', 'Secondary-Server', 'See-Also',
+ 'State', 'Street-Address', 'Tech-Contact', 'Type', 'Updated',
+ 'Updated-By']
+
+ def testAttributes(self):
+ for a in self.attrs:
+ assert(self.db.is_attribute(a))
+ assert(self.db.is_indexed_attr("Host-Name"))
+ assert(not self.db.is_indexed_attr("Phone"))
+ assert(not self.db.is_indexed_attr("foobar"))
+
+ ids = ['local-block-1.a.com', '888.a.com', 'does-not-exist.a.com' ]
+
+ def testFetchObjs(self):
+ objs = self.db.fetch_objects(self.ids)
+ assert(len(objs) == 2)
+ for o in objs:
+ assert(isinstance(o, Rwhois.rwhoisobject))
+ for i in range(len(objs)):
+ self.assertEquals(objs[i].getid().lower(), self.ids[i])
+
+class SearchTest(unittest.TestCase):
+
+ def setUp(self):
+ self.db = MemDB.MemDB()
+ assert(self.db)
+ self.db.init_schema("test_schema")
+ self.db.load_data("test_data")
+
+ attr_search = [ ('domain-name', 'a.com', '333.a.com'),
+ ('ip-network', '10.131.0.0/16', '666.10.0.0.0/8'),
+ ('host-name', 'NS2.A.COM', '5552.a.com'),
+ ('domain-name', 'ns2.a.com', None) ]
+
+ def testAttrSearch(self):
+ for c, k, v in self.attr_search:
+ res = self.db.search_attr(c, k)
+ assert(isinstance(res, MemDB.IndexResult))
+ length = 1
+ if v == None: length = 0
+ self.assertEquals(len(res), length)
+ if length > 0:
+ self.assertEquals(res.list()[0], v)
+
+ nrml_search = [ ('a-net', '666.10.0.0.0/8', 1),
+ ('ns2.a.*', '5552.a.com', 1),
+ ('foo.a.com', None, 0),
+ ('a*', '666.10.0.0.0/8', 3) ]
+
+ def testNormalSearch(self):
+ for k, v, l in self.nrml_search:
+ res = self.db.search_normal(k)
+ assert(isinstance(res, MemDB.IndexResult))
+ self.assertEquals(len(res), l)
+ if l > 0:
+ self.assertEquals(res.list()[0], v)
+ if l > 1:
+ res = self.db.search_normal(k, 1)
+ self.assertEquals(len(res), 1)
+
+
+ cidr_search = [ ('10.0.0.2', '666.10.0.0.0/8', 2),
+ ('192.168.0.0/16**', 'local-block-1.a.com', 1),
+ ('127.0.0.1/26', None, 0) ]
+
+ def testCidrSearch(self):
+ for k, v, l in self.cidr_search:
+ res = self.db.search_cidr(k)
+ assert(isinstance(res, MemDB.IndexResult))
+ self.assertEquals(len(res), l)
+ if l > 0:
+ self.assertEquals(res.list()[0], v)
+ if l > 1:
+ res = self.db.search_cidr(k, 1)
+ self.assertEquals(len(res), 1)
+
+ def testReferralSearch(self):
+ res = self.db.search_referral("fddi.a.com")
+ self.assertEquals(len(res), 1)
+ self.assertEquals(res.list()[0], '888.a.com')
+
+ def testIterator(self):
+ it = self.db.object_iterator()
+ l = [ x for x in it ]
+ self.assertEquals(len(l), 9)
+
+if __name__ == "__main__":
+ unittest.main()
--- /dev/null
+## Data file:
+## This is much like a rwhoisd-1.5.x series data file. It consists
+## of "attribute: value" pairs, separated by lines of "---" or (for
+## this file format) blank lines. Unlike the rwhoisd file format,
+## the Class-Name attribute is required.
+
+## Note: Auth-Area attributes are required and auth-area names
+## either need to be domain names or CIDR formatted network block.
+## Referral objects must have "referred-auth-area" attribute(s) and
+## "referral" attribute(s) for referrals to work at all.
+
+
+ID: 666.10.0.0.0/8
+Class-Name: network
+Auth-Area: 10.0.0.0/8
+Network-Name: A-NET
+IP-Network: 10.0.0.0/8
+Organization:777.a.com
+Tech-Contact:222.a.com
+Admin-Contact:222.a.com
+Created: 19961022
+Updated: 19961023
+Updated-By: hostmaster@a.com
+
+ID: VRSN-BLOCK-7-V6.a.com
+Class-Name: network
+Auth-Area: a.com
+IP-Network: 2001:503:a83e::0/48
+Network-Name: VRSN-BLOCK-7-V6
+Organization: 777.a.com
+Tech-Contact: 222.a.com
+Admin-Contact: 222.a.com
+Created: 20040315
+Updated: 20050704
+Updated-By: hostmaster@a.com
+
+ID: LOCAL-BLOCK-1.a.com
+Class-Name: network
+Auth-Area: a.com
+Network-Block: 192.168.1.0 - 192.168.2.255
+Network-Name: LOCAL-BLOCK-1
+Organization: 777.a.com
+Tech-Contact: 222.a.com
+Admin-Contact: 222.a.com
+Created: 20011224
+Updated: 20030505
+Updated-By: hostmaster@a.com
+
+ID:333.a.com
+Auth-Area:a.com
+Class-Name: domain
+Guardian:444.a.com
+Domain-Name: a.com
+Primary-Server:5551.a.com
+Secondary-Server:5552.a.com
+Organization:777.a.com
+Admin-Contact:222.a.com
+Tech-Contact:222.a.com
+Billing-Contact:222.a.com
+Created:19961022
+Updated:19961023
+Updated-By:hostmaster@a.com
+
+ID:222.a.com
+Auth-Area: a.com
+Class-Name: contact
+Name:Public, John Q.
+Email:johnq@a.com
+Type:I
+First-Name:John
+Last-Name:Public
+Phone:(847)-391-7926
+Fax:(847)-338-0340
+Organization:777.a.com
+See-Also:http://www.a.com/~johnq
+Created:11961022
+Updated:11961023
+Updated-By:hostmaster@a.com
+
+ID:223.a.com
+Auth-Area:a.com
+Class-Name: contact
+Name:Doe, Jane
+Email:janed@a.com
+Type:I
+First-Name:Jane
+Last-Name:Doe
+Last-Name:Doe
+Last-Name:Doe
+Phone:(847)-391-7943
+Fax:(847)-338-0340
+Organization:777.a.com
+Created:11961025
+Updated:11961025
+Updated-By:hostmaster@a.com
+
+ID: 5552.a.com
+Auth-Area: a.com
+Class-Name: host
+Host-Name: ns2.a.com
+IP-Address: 10.0.0.2
+Created: 19961022
+Updated: 19961023
+Updated-By: hostmaster@a.com
+
+ID: 777.a.com
+Auth-Area: a.com
+Class-Name: organization
+Org-Name: A Communications, Inc.
+Street-Address: #600 - 1380 Burrars St.
+City: Vaner
+State: CM
+Postal-Code: V6Z 2H3
+Country-Code: NL
+Phone: (401) 555-6721
+Created: 19961022
+Updated: 19961023
+Updated-By: hostmaster@a.com
+
+ID:888.a.com
+Auth-Area: a.com
+Class-Name: referral
+Guardian:444.a.com
+Referral:rwhois://rwhois.second.a.com:4321/Auth-Area=fddi.a.com
+Organization:777.a.com
+Referred-Auth-Area:fddi.a.com
+Created:19961022
+Updated:19961023
+Updated-By:hostmaster@a.com
git://blacka.com / python-rwhoisd.git / commitdiff