test.py 3.87 KB
Newer Older
Daniel Schreiber's avatar
Daniel Schreiber committed
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
# -*- coding: utf-8 -*-
from __future__ import unicode_literals, print_function

from datetime import date, timedelta
import json
import requests
import sys
import time

def compare_attribute(name, value):
    return lambda attlist: len([e for e in attlist if e['name']==name and e]) == 1


def get_and_compare(url, headers, comp_func):
    tries = 10
    r = None
    while tries > 0:
        r = requests.get(url, headers=headers)
        data_sp = r.json()

        if comp_func(data_sp):
            return True
        tries -= 1
        time.sleep(1)
    print(r.content)
    return False

def main():
    url_sp = "http://localhost:8000/api/res/"
    url_idp = "http://localhost:8001/api/res/"

    token_sp = 'b7bc73430378e67a79a9ca1caf1e39f15944c4d0'
    token_idp = 'b8cbe4bc3fb913b923534faf06f1d0a2a8a3575c'

    headers_sp = {
        'Content-Type': 'application/json',
        "Authorization": "Token {}".format(token_sp)
    }
    headers_idp = {
        'Content-Type': 'application/json',
        "Authorization": "Token {}".format(token_idp)
    }

    today = date.today()

    data = {
        'eppn': 'foo@frms2.edu',
        'sp': '5ebc3b5b-f9c3-4be1-a4aa-d9171c3af6ec',
        'sp_primary_key': 'k',
        'expiry_date': (today + timedelta(days=7)).strftime('%Y-%m-%dT%H:%M:%SZ'),
        'deletion_date': (today + timedelta(days=7+30)).strftime('%Y-%m-%dT%H:%M:%SZ'),
        'setup_token': '1234',
        'attributes': [
            {'name': 'foo', 'value': 'bar'},
            {'name': 'x', 'value': 'y'},
        ],
    }

    r = requests.post(url_sp, json=data, headers=headers_sp)
    if r.status_code != 201:
        print("POST fehlgeschlagen: %s, %s" % (r.content, r.status_code))
        return 1
    print("POST OK, %s" % r.headers['Location'])
    uuid = r.headers['Location'].replace(url_sp, '')
    base, uuid, rest = r.headers['Location'].rsplit('/', 2)
    resource_url_sp = url_sp + uuid + '/'
    resource_url_idp = url_idp + uuid + '/'
    r = requests.get(url_sp, headers=headers_sp)
    print(r.content)
    r = requests.get(url_idp, headers=headers_idp)
    print(r.content)
    
    # test, ob repliziert
    tries = 5
    while tries > 0:
        r = requests.get(resource_url_idp, headers=headers_idp)
        if r.status_code == 200:
            tries = 0
            print("found on remote side\n%s" % r.content)
        else:
            tries -= 1
            time.sleep(2)
            print("remote status code: %s. Trying again %s more times" % (r.status_code, tries))

    r = requests.get(resource_url_idp + 'attributes/', headers=headers_idp)
    data = r.json()
    print(data)
    for entry in data:
        if entry['name'] == 'foo':
            entry['value'] += "_geaendert"
            entry['vec_idp'] += 1
    print("PUT: %s" % data)
    r = requests.put(resource_url_idp + 'attributes/', json=data, headers=headers_idp)
    if r.status_code != 200:
        print("PUT failed: %s\n%s" % (r.status_code, r.content))
        return 1

    if not get_and_compare(resource_url_sp + 'attributes/', headers_sp, compare_attribute('foo', 'bar_geaendert')):
        print("Sync vom IDP zum SP hat nicht geklappt")
        return 1
    print("Sync vom IDP zum SP hat geklappt")
    r = requests.get(resource_url_sp + 'attributes/', headers=headers_sp)
    data = r.json()
    print(data)
    for entry in data:
        if entry['name'] == 'foo':
            entry['value'] = "neu"
            entry['vec_sp'] += 1
    r = requests.put(resource_url_sp + 'attributes/', json=data, headers=headers_sp)
    if not get_and_compare(resource_url_idp + 'attributes/', headers_idp, compare_attribute('foo', 'neu')):
        print("Sync vom SP zum IDP hat nicht geklappt")
        return 1
    print("Sync vom SP zum IDP hat geklappt")
    
    #time.sleep(2)
    #r = requests.get(resource_url_sp + 'attributes/', headers=headers_sp)
    #data_sp = r.json()
    #print(data_sp)
    return 0

sys.exit(main())