docker-atlassian-confluence-server/func-tests/smoketests/tests/test_smoketests.py

import pytest
import requests
import os
import time
import json
from requests.auth import HTTPBasicAuth
import file_helper


def test_create_space(space_key):
    url = f"{REST_API_URL}/space"

    data = {
        "key": space_key,
        "name": "Test new space",
        "description": {
            "plain": {
                "value": "test space generated by api for smoketest",
                "representation": "plain"
            }
        },
        "metadata": {}
    }
    headers = {'Content-Type': 'application/json'}

    r = requests.post(url, data=json.dumps(data), headers=headers, auth=auth)
    assert r.status_code == 200, \
        f'failed to create space {space_key}! status:{r.status_code}'


def test_create_content(space_key, title, content, content_id):
    url = f"{REST_API_URL}/content"
    data = {
        "type": "page",
        "status": "current",
        "title": title,
        "space": {"key": space_key},
        "representation": "storage",
        "body":
            {
                "storage":
                    {
                        "value": f"<p>{content}</p>",
                        "representation": "storage"
                    }
            }
    }
    headers = {'Content-Type': 'application/json'}

    r = requests.post(url, data=json.dumps(data), headers=headers, auth=auth)
    assert r.status_code == 200, \
        f'failed to create page "{title}"! status:{r.status_code}'
    result = json.loads(r.text)
    content_id['val'] = result['id']


def test_search_content(content):
    url = f"{REST_API_URL}/content/search?cql=text~'{content}'"
    attempts = 0
    while True:
        r = requests.get(url, auth=auth)
        if r.status_code == 200:
            if len(json.loads(r.text)['results']) > 0:
                break
        attempts += 1
        if attempts > max_wait_for_response:
            break
        # Add more delay if index still updating and no result is returned
        time.sleep(1)

    assert r.status_code == 200, \
        f'failed to search content! status:{r.status_code}'
    assert len(json.loads(r.text)['results']) > 0, \
        f'failed to find the search term!'


def test_view_content(content_id):
    url = f"{REST_API_URL}/content/{content_id['val']}?expand=body.view"
    r = requests.get(url, auth=auth)
    assert r.status_code == 200, \
        f'failed to view the page! status:{r.status_code}'
    assert len(json.loads(r.text)['body']['view']['value']) > 0, \
        "no content is loaded"


def test_edit_content(content_id):
    url = f"{REST_API_URL}/content/{content_id['val']}"
    data = {
        "version": {
            "number": 2
        },
        "type": "page",
        "title": "New Title",
        "body":
            {
                "storage":
                    {
                        "value": "<p>This page is edited for smoketest!</p>",
                        "representation": "storage"
                    }
            }
    }
    headers = {'Content-Type': 'application/json'}

    r = requests.put(url, data=json.dumps(data), headers=headers, auth=auth)
    assert r.status_code == 200, \
        f'failed to edit page "{title}"! status:{r.status_code}'


def test_add_attachments(content_id, tempfile):
    url = f"{REST_API_URL}/content/{content_id['val']}/child/attachment"
    headers = {'X-Atlassian-Token': 'no-check'}
    # no content-type here!

    r = requests.post(url, headers=headers, files=tempfile, auth=auth)
    assert r.status_code == 200, \
        f'failed to upload the attachment! status:{r.status_code}'


def test_retrieve_attachment(content_id, tempfile):
    url = f"{REST_API_URL}/content/{content_id['val']}/child/attachment"
    headers = {'Content-Type': 'application/json'}
    r = requests.get(url, headers=headers, auth=auth)
    assert r.status_code == 200,\
        f'failed to find the attachments! status:{r.status_code}'
    result = json.loads(r.text)

    assert tempfile['file'][0].endswith(result['results'][0]['title']), \
        "The attachment file is not found!"
    downloadurl = f"{BASE_URL}{result['results'][0]['_links']['download']}"
    attempts = 0
    while True:
        r = requests.get(downloadurl, headers=headers, auth=auth)
        assert r.status_code == 200, "Unable to download the attachment!"
        file_content = open(tempfile['file'][0], 'r').read()
        if file_content == r.text or attempts > max_wait_for_response:
            break
        # makes more delay to upload from last test get in place
        attempts += 1
        time.sleep(1)
    assert file_content == r.text,\
        "The content of downloaded file is not match with the original file"


def test_delete_content(content_id):
    url = f"{REST_API_URL}/content/{content_id['val']}"
    r = requests.delete(url, auth=auth)
    # returns 200(trash) or 204(purge) on success and 404 or 409 if failed
    assert r.status_code == 200 or r.status_code == 204,\
        f'failed to delete the page! status:{r.status_code}'
    confirm_status_404(url)


def test_delete_space(space_key):
    url = f"{REST_API_URL}/space/{space_key}"
    r = requests.delete(url, auth=auth)
    # returns 202 on success and 404 if failed
    assert r.status_code == 202, \
        f'failed to delete the space! status:{r.status_code}'
    confirm_status_404(url)


def confirm_status_404(url):
    attempts = 0
    while True:
        r = requests.get(url, auth=auth)
        if r.status_code == 404 or attempts >= max_wait_for_response:
            break
        attempts += 1
        time.sleep(1)
    assert r.status_code == 404, f'failed to confirm status code 404!'


@pytest.fixture
def space_key():
    return 'testspace01'


@pytest.fixture
def title():
    return 'second page'


@pytest.fixture
def content():
    return 'This page is created for smoketest!'


@pytest.fixture
def tempfile():
    return files


@pytest.fixture
def content_id():
    return os.environ.get('GENERATED_CONTENT_ID')


@pytest.fixture(scope='session')
def content_id():
    return {'val': 0}


# global variablea
BASE_URL = os.environ.get('CONFLUENCE_BASE_URL', "http://localhost:8090")
REST_API_URL = f"{BASE_URL}/rest/api"
password = os.environ.get('CONFLUENCE_ADMIN_PWD', 'admin')
user = os.environ.get('CONFLUENCE_ADMIN', 'admin')
auth = HTTPBasicAuth(user, password)
max_wait_for_response = 30
files = file_helper.create_temp_files()