Go API Documentation

github.com/TykTechnologies/tyk/coprocess/python

No package summary is available.

Package

Files: 0. Third party imports: 0. Imports from organisation: 0. Tests: 0. Benchmarks: 0.


Tests

Files: 2. Third party imports: 0. Imports from organisation: 0. Tests: 3. Benchmarks: 0.

Vars

var pythonBundleWithAuthCheck = func(token1, token2 string) map[string]string {
	middleware := `
from tyk.decorators import *
from gateway import TykGateway as tyk

@Hook
def MyAuthHook(request, session, metadata, spec):
  auth_header = request.get_header('Authorization')
  if auth_header == 'valid_token':
    session.rate = 1000.0
    session.per = 1.0
    session.max_query_depth = 1
    session.quota_max = 1
    session.quota_renewal_rate = 60
    metadata["token"] = "valid_token"
  if auth_header == 'token_without_quota':
    policy_id = request.get_header('Policy')
    session.apply_policy_id = policy_id
    metadata["token"] = "token_without_quota"
  return request, session, metadata
`

	middleware = strings.ReplaceAll(middleware, "valid_token", token1)
	middleware = strings.ReplaceAll(middleware, "token_without_quota", token2)
	checksum := fmt.Sprintf("%x", md5.Sum([]byte(middleware)))

	return map[string]string{
		"manifest.json": fmt.Sprintf(`
		{
		    "file_list": [
		        "middleware.py"
		    ],
		    "custom_middleware": {
		        "driver": "python",
		        "auth_check": {
		            "name": "MyAuthHook"
		        }
		    },
		    "checksum": "%s"
		}`, checksum),
		"middleware.py":	middleware,
	}
}
var pythonBundleWithPostHook = func() map[string]string {
	middleware := `
from tyk.decorators import *
from gateway import TykGateway as tyk
import json

@Hook
def MyPostHook(request, session, spec):
    if "testkey" not in session.metadata.keys():
        request.object.return_overrides.response_code = 400
        request.object.return_overrides.response_error = "'testkey' not found in metadata"
        return request, session
    nested_data = json.loads(session.metadata["testkey"])
    if "nestedkey" not in nested_data:
        request.object.return_overrides.response_code = 400
        request.object.return_overrides.response_error = "'nestedkey' not found in nested metadata"
        return request, session
    if "stringkey" not in session.metadata.keys():
        request.object.return_overrides.response_code = 400
        request.object.return_overrides.response_error = "'stringkey' not found in metadata"
        return request, session
    stringkey = session.metadata["stringkey"]
    if stringkey != "testvalue":
        request.object.return_overrides.response_code = 400
        request.object.return_overrides.response_error = "'stringkey' value doesn't match"
        return request, session	
    return request, session
`
	checksum := fmt.Sprintf("%x", md5.Sum([]byte(middleware)))
	return map[string]string{
		`manifest.json`: fmt.Sprintf(`{
		    "file_list": [
		        "middleware.py"
		    ],
		    "custom_middleware": {
		        "driver": "python",
		        "post": [{
		            "name": "MyPostHook"
		        }]
		    },
			"checksum": "%s"
		}`, checksum),
		"middleware.py":	middleware,
	}
}
var pythonBundleWithPreHook = func() map[string]string {
	middleware := `
from tyk.decorators import *
from gateway import TykGateway as tyk

@Hook
def MyPreHook(request, session, metadata, spec):
    content_type = request.get_header("Content-Type")
    if "json" in content_type:
      if len(request.object.raw_body) <= 0:
        request.object.return_overrides.response_code = 400
        request.object.return_overrides.response_error = "Raw body field is empty"
        return request, session, metadata
      if "{}" not in request.object.body:
        request.object.return_overrides.response_code = 400
        request.object.return_overrides.response_error = "Body field doesn't match"
        return request, session, metadata
    if "multipart" in content_type:
      if len(request.object.body) != 0:
        request.object.return_overrides.response_code = 400
        request.object.return_overrides.response_error = "Body field isn't empty"
      if len(request.object.raw_body) <= 0:
        request.object.return_overrides.response_code = 400
        request.object.return_overrides.response_error = "Raw body field is empty"
    return request, session, metadata
`
	checksum := fmt.Sprintf("%x", md5.Sum([]byte(middleware)))
	return map[string]string{
		`manifest.json`: fmt.Sprintf(`{
		    "file_list": [
		        "middleware.py"
		    ],
		    "custom_middleware": {
		        "driver": "python",
		        "pre": [{
		            "name": "MyPreHook"
		        }]
		    },
			"checksum": "%s"
		}`, checksum),
		"middleware.py":	middleware,
	}
}
var pythonBundleWithResponseHook = func() map[string]string {
	middleware := `
from tyk.decorators import *
from gateway import TykGateway as tyk

@Hook
def MyResponseHook(request, response, session, metadata, spec):
  response.raw_body = b'newbody'
  return response
`
	checksum := fmt.Sprintf("%x", md5.Sum([]byte(middleware)))
	return map[string]string{
		`manifest.json`: fmt.Sprintf(`
		{
		    "file_list": [
		        "middleware.py"
		    ],
		    "custom_middleware": {
		        "driver": "python",
		        "response": [{
		            "name": "MyResponseHook"
		        }]
		    },
			"checksum": "%s"
		}
`, checksum),
		"middleware.py":	middleware,
	}
}
var pythonIDExtractorFormValue = func(token string) map[string]string {
	middleware := `
import time
from tyk.decorators import *
from gateway import TykGateway as tyk
from urllib import parse

counter = 0

@Hook
def MyAuthHook(request, session, metadata, spec):
    global counter
    counter = counter + 1
    auth_param = parse.parse_qs(request.object.body)["auth"]
    if auth_param and auth_param[0] == 'valid_token' and counter < 2:
        session.rate = 1000.0
        session.per = 1.0
        session.id_extractor_deadline = int(time.time()) + 60
        metadata["token"] = "valid_token"
    return request, session, metadata
`

	middleware = strings.ReplaceAll(middleware, "valid_token", token)
	checksum := fmt.Sprintf("%x", md5.Sum([]byte(middleware)))

	return map[string]string{
		"manifest.json": fmt.Sprintf(`
		{
		    "file_list": [
		        "middleware.py"
		    ],
		    "custom_middleware": {
		        "driver": "python",
		        "auth_check": {
		            "name": "MyAuthHook"
		        },
		        "id_extractor": {
		        	"extract_from": "form",
		        	"extract_with": "value",
		        	"extractor_config": {
		        		"param_name": "auth"
		        	}
		        }
		    },
			"checksum":"%s"
		}
	`, checksum),
		"middleware.py":	middleware,
	}
}
var pythonIDExtractorHeaderRegex = func(token string) map[string]string {
	middleware := `
import time
from tyk.decorators import *
from gateway import TykGateway as tyk

counter = 0

@Hook
def MyAuthHook(request, session, metadata, spec):
    print("MyAuthHook3 is called")
    global counter
    counter = counter + 1
    _, auth_header = request.get_header('Authorization').split('-')
    if auth_header and auth_header == '12345' and counter < 2:
        session.rate = 1000.0
        session.per = 1.0
        session.id_extractor_deadline = int(time.time()) + 60
        metadata["token"] = "valid_token"
    return request, session, metadata
	`

	middleware = strings.ReplaceAll(middleware, "valid_token", token)
	checksum := fmt.Sprintf("%x", md5.Sum([]byte(middleware)))

	return map[string]string{
		"manifest.json": fmt.Sprintf(`
		{
		    "file_list": [
		        "middleware.py"
		    ],
		    "custom_middleware": {
		        "driver": "python",
		        "auth_check": {
		            "name": "MyAuthHook"
		        },
		        "id_extractor": {
		        	"extract_from": "header",
		        	"extract_with": "regex",
		        	"extractor_config": {
		        		"header_name": "Authorization",
						"regex_expression": "[0-9]+",
						"regex_match_index": 0
		        	}
		        }
		    },
			"checksum":"%s"
		}
	`, checksum),
		"middleware.py":	middleware,
	}
}
var pythonIDExtractorHeaderValue = func(token string) map[string]string {
	middleware := `
import time
from tyk.decorators import *
from gateway import TykGateway as tyk

counter = 0

@Hook
def MyAuthHook(request, session, metadata, spec):
    global counter
    counter = counter + 1
    auth_header = request.get_header('Authorization')
    if auth_header == 'valid_token' and counter < 2:
        session.rate = 1000.0
        session.per = 1.0
        session.id_extractor_deadline = int(time.time()) + 60
        metadata["token"] = "valid_token"
    return request, session, metadata
	`

	middleware = strings.ReplaceAll(middleware, "valid_token", token)
	checksum := fmt.Sprintf("%x", md5.Sum([]byte(middleware)))

	return map[string]string{
		"manifest.json": fmt.Sprintf(`
		{
		    "file_list": [
		        "middleware.py"
		    ],
		    "custom_middleware": {
		        "driver": "python",
		        "auth_check": {
		            "name": "MyAuthHook"
		        },
		        "id_extractor": {
		        	"extract_from": "header",
		        	"extract_with": "value",
		        	"extractor_config": {
		        		"header_name": "Authorization"
		        	}
		        }
		    },
			"checksum":"%s"
		}
	`, checksum),
		"middleware.py":	middleware,
	}
}
var pythonPostRequestTransform = func() map[string]string {
	middleware := `
from tyk.decorators import *
from gateway import TykGateway as tyk
import json

@Hook
def MyPostHook(request, session, spec):
	if request.object.url == "/test2":
		if request.object.method != "POST":
			request.object.return_overrides.response_code = 500
			request.object.return_overrides.response_error = "'invalid method type'"
			return request, session
		request.object.url = "tyk://test-api-2/newpath"
		request.object.method = "GET"

	return request , session
`
	checksum := fmt.Sprintf("%x", md5.Sum([]byte(middleware)))
	return map[string]string{
		`manifest.json`: fmt.Sprintf(`{
		    "file_list": [
		        "middleware.py"
		    ],
		    "custom_middleware": {
		        "driver": "python",
		        "post": [{
		            "name": "MyPostHook"
		        }]
		    },
			"checksum": "%s"
		}`, checksum),
		"middleware.py":	middleware,
	}
}

Test functions

TestMain

References: context.Background, gateway.InitTestMain, os.Exit.

TestPythonBundles

References: apidef.MethodTransformMeta, apidef.URLRewriteMeta, bytes.Buffer, gateway.APISpec, gateway.CreateSession, gateway.GenerateTestBinaryData, http.MethodDelete, http.MethodGet, http.MethodPost, http.StatusForbidden, http.StatusOK, multipart.NewWriter, test.TestCase, testing.T, user.AccessDefinition, user.Policy, user.SessionState, uuid.NewHex.

TestValueExtractorHeaderSource

References: config.CoProcessConfig, gateway.APISpec, gateway.StartTest, gateway.TestConfig, http.StatusForbidden, http.StatusOK, test.TestCase, testing.T, url.Values, uuid.NewHex.