REST case for neo4j

rest_graph.py

__author__ = 'jude'
import requests
import base64
import unittest
import json
import pprint
def payload(user, passwd):
    return base64.b64encode("{0}:{1}".format(user,passwd))

class AuthCase(unittest.TestCase):
    def testAuth(self):
        url =  "http://121.201.29.89:7474/user/neo4j"
        head = {'content-type': 'application/json',
                'Accept': 'application/json; charset=UTF-8',
                'Authorization':'Base %s' % payload('neo4j','123456') }

        r = requests.get(url,headers=head)
        self.assertEquals(r.status_code,200)
        # pprint.pprint( r.headers)

class RestCase(unittest.TestCase):
    def setUp(self):
        self.header = {'content-type': 'application/json',
                     'Accept': 'application/json; charset=UTF-8',
                    'Authorization':'Base %s' % payload('neo4j','123456')}

    def testCreate(self):
        url =  "http://121.201.29.89:7474/db/data/transaction/commit"
        data = {
        "statements" : [{
        "statement" : "CREATE (n:FOO) RETURN id(n)"
        }]
       }
        r = requests.post(url,data=json.dumps(data) ,headers=self.header)
        self.assertEquals(r.status_code, 200)
        self.assertEquals(r.headers['Content-Type'], 'application/json')
        # print r.headers
        pprint.pprint(r.content)

    def testMultiStatement(self):
        url = "http://121.201.29.89:7474/db/data/transaction/commit"
        data = {
              "statements" : [ {
                "statement" : "CREATE (n:FOO) RETURN id(n)"
              }, {
                "statement" : "CREATE (n {props}) RETURN n",
                "parameters" : {
                  "props" : {
                    "name" : "My Node"
                  }
                }
              } ]
            }
        r = requests.post(url,data=json.dumps(data) ,headers=self.header)
        self.assertEquals(r.status_code, 200)
        self.assertEquals(r.headers['Content-Type'], 'application/json')
        print r.headers
        pprint.pprint(r.content)

    def testTransaction(self):
        url = "http://121.201.29.89:7474/db/data/transaction"
        data = {
              "statements" : [ {
                "statement" : "CREATE (n:FOO {props}) RETURN n",
                "parameters" : {
                  "props" : {
                    "name" : "My Node"
                  }
                }
              } ]
            }
        r = requests.post(url,data=json.dumps(data) ,headers=self.header)
        self.assertEquals(r.status_code, 201)
        self.assertEquals(r.headers['Content-Type'], 'application/json')
        # print r.headers
        pprint.pprint(r.content)

    def testCreateNode(self):
        url = "http://121.201.29.89:7474/db/data/node"

        r = requests.post(url ,headers=self.header)
        self.assertEquals(r.status_code, 201)
        self.assertEquals(r.headers['Content-Type'], 'application/json; charset=UTF-8')
        # print r.headers['Location']
        print r.content

def testGetNodes(self):
        url = "http://121.201.29.89:7474/db/data/transaction/commit"
        data = {
        "statements" : [{
        "statement" : "MATCH ()-[r:ENT_BH_TO_BDB_BH]->() RETURN r LIMIT 25",
        "resultDataContents" : [ "row", "graph" ]
        }],

       }
        r = requests.post(url,data=json.dumps(data) ,headers=self.header)
        self.assertEquals(r.status_code, 200)
        self.assertEquals(r.headers['Content-Type'], 'application/json')
        self.assertEquals(type(r.content), type(''))
        print json.loads(r.content)

if __name__ == "__main__":
    unittest.main()