text:
/* eslint-disable no-unused-expressions */

const chai = require('chai')
chai.use(require('chai-http'))
const expect = chai.expect

const constants = require('../constants.js')
const app = require('../../../src/index.js')
const helpers = require('../helpers.js')

const shortName = 'win_5'

describe('Text PUT CVE-ID/:id', () => {
  let cveId
  before(async () => {
    cveId = await helpers.cveIdReserveHelper(1, '2023', shortName, 'non-sequential')
  })
  context('State parameter Tests', () => {
    it('Endpoint should return a 400 when state org is set to published', async () => {
      await chai.request(app)
        .put(`/api/cve-id/${cveId}?state=PUBLISHED`)
        .set(constants.headers)
        .then((res, err) => {
          expect(err).to.be.undefined
          expect(res).to.have.status(400)
        })
    })
    it('Endpoint should allow the state parameter to be set to reserved', async () => {
      await chai.request(app)
        .put(`/api/cve-id/${cveId}?state=REJECTED`)
        .set(constants.headers)
        .then((res, err) => {
          expect(err).to.be.undefined
          expect(res).to.have.status(200)
        })
    })
    it('Endpoint should still not allow a REJECTED endpoint to be set to published', async () => {
      await chai.request(app)
        .put(`/api/cve-id/${cveId}?state=PUBLISHED`)
        .set(constants.headers)
        .then((res, err) => {
          expect(err).to.be.undefined
          expect(res).to.have.status(400)
        })
    })
  })
})


text:
/* eslint-disable no-unused-expressions */

const chai = require('chai')
chai.use(require('chai-http'))

const expect = chai.expect

const constants = require('../constants.js')
const app = require('../../../src/index.js')

const requestLength = 10

describe('Testing Reserve CVE-ID Endpoints', () => {
  // beforeEach(() => { })
  context('Positive Tests', () => {
    it('Should return 200 and have correct number of cve-id results for a successful non-squential cve-id reservation ', (done) => {
      chai.request(app)
        .post(`/api/cve-id?amount=${requestLength}&cve_year=2023&short_name=mitre&batch_type=non-sequential`)
        .set(constants.headers)
        .end((err, res) => {
          expect(err).to.be.null
          expect(res.body).to.have.property('cve_ids')
          expect(res.body.cve_ids).to.have.length(requestLength)
          expect(res).to.have.status(200)
          done()
        })
    })
    it('Should return 200 and have correct number of cve-id results in sequential order for a successful sequential cve-id reserveration ', (done) => {
      chai.request(app)
        .post('/api/cve-id?amount=10&cve_year=2023&short_name=mitre&batch_type=sequential')
        .set(constants.headers)
        .end((err, res) => {
          expect(err).to.be.null
          expect(res.body.cve_ids).to.have.length(10)
          expect(res).to.have.status(200)
          const onlyCveIdNumbers = res.body.cve_ids.map((element) => { return parseInt(element.cve_id.split('-')[2]) })
          const cveNumbersInOrder = onlyCveIdNumbers.every((value, index) => {
            if (index === 0) {
              return true
            } else {
              return value >= onlyCveIdNumbers[index - 1]
            }
          })
          expect(cveNumbersInOrder).to.be.true
          done()
        })
    })
  })
  context('Negative Tests', () => {
    it('Should return 403 when a request is by a user to reserve cve-ids outside of their organization ', (done) => {
      chai.request(app)
        .post('/api/cve-id?amount=10&cve_year=2023&short_name=mitre&batch_type=sequential')
        .set(constants.nonSecretariatUserHeaders)
        .end((err, res) => {
          expect(err).to.be.null
          expect(res).to.have.status(403)
          done()
        })
    })
  })

  // afterEach(() => { })
})

repo_name: CVEProject/cve-services

repo_language: JavaScript

file_name: Dockerfile

mime_type: text/plain


text:
/* eslint-disable no-unused-expressions */

const chai = require('chai')
chai.use(require('chai-http'))
const _ = require('lodash')

const expect = chai.expect

const constants = require('../constants.js')
const app = require('../../../src/index.js')

describe('Testing Get CVE-ID endpoint', () => {
  // TODO: Update this test to dynamically calculate reserved count.
  const RESESRVED_COUNT = 120
  const YEAR_COUNT = 10
  const PUB_YEAR_COUNT = 4
  const TIME_WINDOW_COUNT = 40

  context('Positive Tests', () => {
    it('Get CVE-ID should return everything when no parameters are specifed', async () => {
      await chai.request(app)
        .get('/api/cve-id')
        .set(constants.headers)
        .then((res, err) => {
          expect(err).to.be.undefined
          expect(res).to.have.status(200)
        })
    })
    it('Get CVE-ID should return an empty array when time modified is set to a very far future date', async () => {
      await chai.request(app)
        .get('/api/cve-id?time_modified.gt=2100-01-01T00:00:00')
        .set(constants.headers)
        .then((res, err) => {
          expect(err).to.be.undefined
          expect(res).to.have.status(200)
          expect(res.body.cve_ids).to.have.length(0)
        })
    })
    // Need a better way to test each individual cve-id's time_modified
    it('Get all CVE-IDs modified within a given timeframe', async () => {
      await chai.request(app)
        .get('/api/cve-id?time_modified.gt=2021-05-11T15:05:20.093Z&time_modified.lt=2021-05-11T15:07:00.093Z')
        .set(constants.headers)
        .then((res, err) => {
          expect(err).to.be.undefined
          expect(res).to.have.status(200)
          expect(res.body.cve_ids).to.have.length(TIME_WINDOW_COUNT)
        })
    })
    it('Get all CVE-IDs in the RESERVED state', async () => {
      await chai.request(app)
        .get('/api/cve-id?state=RESERVED')
        .set(constants.headers)
        .then((res, err) => {
          expect(err).to.be.undefined
          expect(res).to.have.status(200)
          expect(_.every(res.body.cve_ids, { state: 'RESERVED' })).to.be.true
          expect(res.body.cve_ids).to.have.length(RESESRVED_COUNT)
        })
    })
    it('Get all CVE-IDs in the PUBLISHED state', async () => {
      await chai.request(app)
        .get('/api/cve-id?state=PUBLISHED')
        .set(constants.headers)
        .then((res, err) => {
          expect(err).to.be.undefined
          expect(res).to.have.status(200)
          expect(_.every(res.body.cve_ids, { state: 'PUBLISHED' })).to.be.true
        })
    })
    it('Get all CVE-IDs in the REJECTED state', async () => {
      await chai.request(app)
        .get('/api/cve-id?state=REJECTED')
        .set(constants.headers)
        .then((res, err) => {
          expect(err).to.be.undefined
          expect(res).to.have.status(200)
          expect(_.every(res.body.cve_ids, { state: 'REJECTED' })).to.be.true
        })
    })
    it('Get all CVE-IDs with cve_id_year 1999', async () => {
      await chai.request(app)
        .get('/api/cve-id?cve_id_year=1999')
        .set(constants.headers)
        .then((res, err) => {
          expect(err).to.be.undefined
          expect(res).to.have.status(200)
          expect(_.every(res.body.cve_ids, { cve_year: '1999' })).to.be.true
          expect(res.body.cve_ids).to.have.length(YEAR_COUNT)
        })
    })
    it('Get all CVE-IDs with cve_id_year 1999 with state PUBLISHED', async () => {
      await chai.request(app)
        .get('/api/cve-id?cve_id_year=1999&state=PUBLISHED')
        .set(constants.headers)
        .then((res, err) => {
          expect(err).to.be.undefined
          expect(res).to.have.status(200)
          expect(_.every(res.body.cve_ids, { cve_year: '1999', state: 'PUBLISHED' })).to.be.true
          expect(res.body.cve_ids).to.have.length(PUB_YEAR_COUNT)
        })
    })
  })
})


text:
/* eslint-disable no-unused-expressions */

const chai = require('chai')
chai.use(require('chai-http'))

const expect = chai.expect

const constants = require('../constants.js')
const app = require('../../../src/index.js')

const body = {
  username: 'adpUser',
  active: 'true',
  name: {
    first: 'TestCnaAdmin',
    last: 'test',
    middle: 'N',
    suffix: 'I'
  },
  authority: {
    active_roles: ['Admin']
  }
}
describe('Testing create user endpoint', () => {
  it('Should return 200 and new user', (done) => {
    chai.request(app)
      .post('/api/org/range_4/user')
      .set(constants.headers)
      .send(body)
      .end((err, res) => {
        expect(err).to.be.null
        expect(res.body).to.have.property('created')
        expect(res.body.created.username).to.equal(body.username)
        expect(res).to.have.status(200)
        done()
      })
  })
})


text:
/* eslint-disable no-unused-expressions */
const chai = require('chai')
chai.use(require('chai-http'))
const expect = chai.expect
const { faker } = require('@faker-js/faker')

const constants = require('../constants.js')
const app = require('../../../src/index.js')
const getConstants = require('../../../src/constants').getConstants
const argon2 = require('argon2')
const _ = require('lodash')
const User = require('../../../src/model/user')
const cryptoRandomString = require('crypto-random-string')

const shortName = { shortname: 'win_5' }

describe('Testing user post endpoint', () => {
  let orgUuid
  context('Positive Tests', () => {
    it('Allows creation of user', async () => {
      await chai.request(app)
        .post('/api/org/win_5/user')
        .set({ ...constants.headers, ...shortName })
        .send({
          username: 'fakeuser999',
          name: {
            first: 'Dave',
            last: 'FakeLastName',
            middle: 'Cool',
            suffix: 'Mr.'
          },
          authority: {
            active_roles: [
              'ADMIN'
            ]
          }
        })
        .then((res, err) => {
          expect(err).to.be.undefined
          orgUuid = res.body.created.org_UUID
        })
    })
  })
  context('Negitive Test', () => {
    it('Fails creation of user for bad long first name', async () => {
      await chai.request(app)
        .post('/api/org/win_5/user')
        .set({ ...constants.headers, ...shortName })
        .send({
          username: 'fakeuser9999',
          name: {
            first: 'VerylongnmVerylongnmVerylongnmVerylongnmVerylongnmVerylongnmVerylongnmVerylongnmVerylongnmVerylongnm1',
            last: 'FakeLastName',
            middle: 'Cool',
            suffix: 'Mr.'
          },
          authority: {
            active_roles: [
              'ADMIN'
            ]
          }
        })
        .then((res, err) => {
          expect(res).to.have.status(400)
          expect(_.some(res.body.details, { msg: 'Invalid name.first. Name must be between 1 and 100 characters in length.' })).to.be.true
          expect(err).to.be.undefined
        })
    })
    it('Fails creation of user for bad long last name', async () => {
      await chai.request(app)
        .post('/api/org/win_5/user')
        .set({ ...constants.headers, ...shortName })
        .send({
          username: 'fakeuser1000',
          name: {
            first: 'FirstName',
            last: 'VerylongnmVerylongnmVerylongnmVerylongnmVerylongnmVerylongnmVerylongnmVerylongnmVerylongnmVerylongnm1',
            middle: 'Cool',
            suffix: 'Mr.'
          },
          authority: {
            active_roles: [
              'ADMIN'
            ]
          }
        })
        .then((res, err) => {
          expect(res).to.have.status(400)
          expect(_.some(res.body.details, { msg: 'Invalid name.last. Name must be between 1 and 100 characters in length.' })).to.be.true
          expect(err).to.be.undefined
        })
    })
    it('Fails creation of user for bad long middle name', async () => {
      await chai.request(app)
        .post('/api/org/win_5/user')
        .set({ ...constants.headers, ...shortName })
        .send({
          username: 'fakeuser1001',
          name: {
            first: 'FirstName',
            last: 'LastName',
            middle: 'VerylongnmVerylongnmVerylongnmVerylongnmVerylongnmVerylongnmVerylongnmVerylongnmVerylongnmVerylongnm1',
            suffix: 'Mr.'
          },
          authority: {
            active_roles: [
              'ADMIN'
            ]
          }
        })
        .then((res, err) => {
          expect(res).to.have.status(400)
          expect(_.some(res.body.details, { msg: 'Invalid name.middle. Name must be between 1 and 100 characters in length.' })).to.be.true
          expect(err).to.be.undefined
        })
    })
    it('Fails creation of user for bad long suffix name', async () => {
      await chai.request(app)
        .post('/api/org/win_5/user')
        .set({ ...constants.headers, ...shortName })
        .send({
          username: 'fakeuser1002',
          name: {
            first: 'FirstName',
            last: 'LastName',
            middle: 'MiddleName',
            suffix: 'VerylongnmVerylongnmVerylongnmVerylongnmVerylongnmVerylongnmVerylongnmVerylongnmVerylongnmVerylongnm1'
          },
          authority: {
            active_roles: [
              'ADMIN'
            ]
          }
        })
        .then((res, err) => {
          expect(res).to.have.status(400)
          expect(_.some(res.body.details, { msg: 'Invalid name.suffix. Name must be between 1 and 100 characters in length.' })).to.be.true
          expect(err).to.be.undefined
        })
    })
    it('Fails creation of user for trying to add the 101th user', async () => {
      const numberOfUsers = await User.where({ org_UUID: orgUuid }).countDocuments().exec()
      for (let i = 0; i < (100 - numberOfUsers); i++) {
        const newUser = new User()

        newUser.name.first = faker.name.firstName()
        newUser.name.last = faker.name.lastName()
        newUser.username = faker.internet.userName({ firstName: newUser.name.first, lastName: newUser.name.last })
        newUser.org_UUID = orgUuid
        newUser.UUID = faker.datatype.uuid()
        const randomKey = cryptoRandomString({ length: getConstants().CRYPTO_RANDOM_STRING_LENGTH })
        newUser.secret = await argon2.hash(randomKey)

        newUser.authority = {
          active_roles: [
            'ADMIN'
          ]
        }

        await User.findOneAndUpdate().byUserNameAndOrgUUID(newUser.userName, newUser.org_UUID).updateOne(newUser).setOptions({ upsert: true })
      }

      await chai.request(app)
        .post('/api/org/win_5/user')
        .set({ ...constants.headers, ...shortName })
        .send({
          username: 'Fake101User',
          name: {
            first: 'Dave',
            last: 'FakeLastName',
            middle: 'Cool',
            suffix: 'Mr.'
          },
          authority: {
            active_roles: [
              'ADMIN'
            ]
          }
        })
        .then((res, err) => {
          expect(err).to.be.undefined
          expect(res).to.have.status(400)
          expect(res.body.error).to.contain('NUMBER_OF_USERS_IN_ORG_LIMIT_REACHED')
        })
    })
  })
})


text:
# Upload 5.0 records
# For now, update directories appropriately
import sys
import json
import getopt
from cve4to5up import CVE_Convert

def main(argv):
    try:
        opts, args = getopt.getopt(argv, "hi:u:d:", ["icvefile=","iuserfile=","ocvedir="])
    except getopt.GetopError:
        print ('USAGE python UpdateCveAssignerv5.py -i <inputcvefile> -u <inputuserfile> -d <CvesOutputDir>')
        sys.exit(2)
    for opt, arg in opts:
        if opt == '-h':
            sys.exit()
        elif opt in ("-i", "--icvefile"):
            inputCveFile = arg
        elif opt in ("-u", "--iuserfile"):
            inputUserFile = arg
        elif opt in ("-d", "--ocvedir"):
            outputpath = arg
    with open(inputCveFile, 'r+') as cves, open(inputUserFile) as users:
        cve_list = json.load(cves)
        user_list = json.load(users)
        for cve in cve_list:
            assigner = cve['cve']['CVE_data_meta']['ASSIGNER']
            id = cve['cve']['CVE_data_meta']['ID']
            # mitre org
            new_assigner = "9845b82e-47a9-4fe7-8908-68646f952267"
            for user in user_list:
                if assigner == user['username']:
                    new_assigner = user['org_UUID']

            with open('temp.json', 'a+') as f:
                f.seek(0)
                f.truncate()
                json.dump(cve['cve'], f)

            CVE_Convert('temp.json', 'test-script-results/', new_assigner)
            with open(f'{outputpath}/{id}.json') as f:
                json_data = json.load(f)
                cve['cve'] = json_data
        cves.seek(0)
        json.dump(cve_list, cves, indent=4)
        cves.truncate()

if __name__ == "__main__":
   main(sys.argv[1:])


text:
var chai = require('chai')
var chaiHttp = require('chai-http')

chai.use(chaiHttp)
chai.should()

describe('API Health Check: ', () => {
  it('Api should be running', function (done) {
    chai.request('http://localhost:3000')
      .get('/health-check')
      .end((err, res) => {
        if (err) {
          console.log(err.stack)
        }

        res.should.have.status(200)
        res.should.be.json()
        res.body.should.be.a('object')
        res.body.should.have.property('isHealthy')
        res.body.isHealthy.should.equal(true)
        done()
      })
  })
})


text:
var chai = require('chai')
var chaiHttp = require('chai-http')
chai.use(require('chai-uuid'))
chai.use(chaiHttp)
chai.should()

var { TEST_CNA_UUID, TEST_USER_PUBLICKEY } = require('../test/constants')

describe('CNA Operations:', () => {
  it('Should Register User', function (done) {
    chai.request('http://localhost:3000')
      .post('/register-user')
      .set(
        'signature',
        'e3fac5295e9b95746b51d849bf6ede74b5df981e18048e48fb2c14c79852c488c9f8bf6ed50be0188cf89f07a47ceeb777f450b5ed898432b90cecca8ed25ae199ce95ee811d5e69b2b560b109016e43be6f1fad43faf4b1179276d0fa41125b990f8f7caf42ae20942876caff65fc8cf5e4759c9fe515ac45a8deb7ad1eb67e'
      )
      .send({
        payload: {
          userPublicKey: TEST_USER_PUBLICKEY,
          userType: 'VENDOR',
          userHardQuota: 100,
          userSoftQuota: 50,
          userDescription: 'user description',
          userGithubID: 'rotsen91',
          userFName: 'Nestor',
          userLName: 'Mora',
          userEmail: 'nestor@test.com',
          userURL: 'Optional userURL',
          userGPGKeys: 'Optional userGPGKeys'
        },
        userUUID: TEST_CNA_UUID
      })
      .end((err, res) => {
        if (err) {
          console.log(err.stack)
        }

        res.should.have.status(200)
        res.should.be.json()
        res.body.should.be.a('object')
        res.body.should.have.property('userUUID')
        res.body.should.have.property('message')
        res.body.message.should.equal('user created')
        res.body.userUUID.should.be.a.uuid('v4')
        done()
      })
  })
})


text:
var chai = require('chai')
var chaiHttp = require('chai-http')
var assertArrays = require('chai-arrays')
chai.use(assertArrays)
chai.use(require('chai-like'))
chai.use(require('chai-things')) // Don't swap these two
chai.use(require('chai-string'))
chai.use(chaiHttp)
chai.should()

var { TEST_USER_UUID, TEST_CVE_STATUS, TEST_CVE_ID } = require('../test/constants')

describe('User Request CVE ID Operations:', () => {
  it('Should Request 2 CVE IDs', function (done) {
    chai.request('http://localhost:3000')
      .post('/get-cve-id')
      .set(
        'signature',
        '0303e5dd1a65426e3565614692d2f6cd8fc279dfbb1c7148104431645641f5e084155cce00879252488d0d68c483996f36701345068b0fa0b8cc0509cf00aa068adb65fa43b871e6a026a936230d10364f286cec22334868cba5c1e8e8b0ac8be4f28ecbe9f4b99ba8803fa04570fae1dd5b16d7c90cc066ac461fdc84980f00'
      )
      .send({ payload: { count: '2' }, userUUID: TEST_USER_UUID })
      .end((err, res) => {
        if (err) {
          console.log(err.stack)
        }

        res.should.have.status(200)
        res.should.be.an('array')
        res.body.should.be.ofSize(2)
        res.body.should.be.an('array').that.contains.something.like({ status: 'Disputed' })
        done()
      })
  })

  it('Should Overide Default Status', function (done) {
    chai.request('http://localhost:3000')
      .post('/get-cve-id')
      .set(
        'signature',
        '485acc0cd2efb4c9f055614f578c1ce080626521f887bb188cd65a0dee17159e217c2c0c5e1add4902d05e5a0b575953a79ea030522c411b77b119152654b4009a55fc7117299463f14f11edf142e10dced4794491a2e138bd1e5b474fb0c60f60a7f8a5a2fb921f67f81de718b441d5ebe87d77c3fec9a117fd40b30e6d8d9d'
      )
      .send({ payload: { status: 'Free' }, userUUID: TEST_USER_UUID })
      .end((err, res) => {
        if (err) {
          console.log(err.stack)
        }

        res.should.have.status(200)
        res.should.be.an('array')
        res.body.should.be.ofSize(1)
        res.body.should.be.an('array').that.contains.something.like({ status: 'Free' })
        done()
      })
  })

  it('Should Overide Current Year', function (done) {
    chai.request('http://localhost:3000')
      .post('/get-cve-id')
      .set(
        'signature',
        '302ce4dc7a2ffe20fae3b0a0874f130cac0e82f16238170b2f7645298aa372f3fcb5a677cf4bb07a7130c5fa43718f6a4493aeaef2a5e3b588119d123ee8a504c73ae3ab7255adf3cec0c4f7b0459aaa80d3ee8e346072d39dd2a916636733ec3f4e7616ce762944ace8dd90922cd82e9c017282cd676c532f7a3704c4bbd8b9'
      )
      .send({ payload: { year: '1991' }, userUUID: TEST_USER_UUID })
      .end((err, res) => {
        if (err) {
          console.log(err.stack)
        }

        res.should.have.status(200)
        res.should.be.an('array')
        res.body.should.be.ofSize(1)
        res.body[0].cve_id.should.have.entriesCount('1991', 1)
        done()
      })
  })
})

describe('User Request CVE Info:', () => {
  it('Should Request Info from CVE with ID', function (done) {
    chai.request('http://localhost:3000')
      .post('/get-cvs-info')
      .set(
        'signature',
        '0452d6c781d07c2346766cc88984fe93fc2061b1d7846c949c35ea76aca85274cab23b2d7cc789b5cd6191ef3235c26ab434c5e36f7c60d023da27bec6e203863eeaff9e6fa949193c15989c0e3215c443fcd776e6cd4c7ef03df7cf5304092e19bbabd67dbdc8d797e1831d197cc878c865e756017f2b18f79bdca5dff9c2e3'
      )
      .send({ payload: { ids: [TEST_CVE_ID] }, userUUID: TEST_USER_UUID })
      .end((err, res) => {
        if (err) {
          console.log(err.stack)
        }

        res.should.have.status(200)
        res.should.be.an('array')
        res.body.should.be.an('array').that.contains.something.like({ cve_id: TEST_CVE_ID })
        done()
      })
  })

  it('Should Request Info from CVE with Status Filter', function (done) {
    chai.request('http://localhost:3000')
      .post('/get-cvs-info')
      .set(
        'signature',
        '4db4d366e23992f30f2770c8f1f8fc532d740ba7228561ef9e4853f80bd9252cbcc76432827277d486d9b608fa7ba946ed244a86ab12d11b67f3342b8ecde8b7a33882db120e2a4d8a01f057449dfa81fb5007783311474c6f8e79ca15596801ff45b88cb7c13cf39b524abfaafbd4ce6c6d86f953ff94239fe017d93c61ff1c'
      )
      .send({ payload: { filter: TEST_CVE_STATUS }, userUUID: TEST_USER_UUID })
      .end((err, res) => {
        if (err) {
          console.log(err.stack)
        }

        res.should.have.status(200)
        res.should.be.an('array')
        res.body.should.be.an('array').that.contains.something.like({ status: TEST_CVE_STATUS })
        done()
      })
  })
})

describe('User Modify CVE Status:', () => {
  it('Should Set CVE With Populated Status ', function (done) {
    chai.request('http://localhost:3000')
      .post('/set-cve-status')
      .set(
        'signature',
        '1f9089ecf145e254c427c7856c7a5c2141f19176777530142398fb67ebb30c5e98d472320e443cb6bc0c93a2c1606a48f0a9557c528e5f6c98397e61191ad11a64aa81b82a6a8af86ba5898d1feafc4f1eecc860808e359656686e17a127fc4bf9207f2416b333e9e95fdf9aebf2c42a47338bb8c53f738fbeb611f2ff990fa9'
      )
      .send({
        payload: {
          ids: [TEST_CVE_ID],
          status: 'Populated'
        },
        userUUID: TEST_USER_UUID
      })
      .end((err, res) => {
        if (err) {
          console.log(err.stack)
        }

        res.should.have.status(200)
        res.body.should.be.a('object')
        res.body.should.have.property('Success')
        res.body.Success.should.equal(true)
        done()
      })
  })
})


text:
# CVE Services Endpoint Testing

**Warning**: do not point these tests at production!!!

This portion of the repository contains HTTP tests written using Python `requests`, and `pytest`. They can be run against any instantiation of the REST API by updating `test-http/docker/.docker-env` accordingly. The REST API must have a backing MongoDb database. Tests should not be run against production, since they do update and delete data. Some tests assume that the MITRE organization exists and is the secretariat.

## Usage

The following will run tests against an existing instance of the CPS and services API, given that the docker environment manifest is updated with valid credentials.

```sh
cd test-http

# Copy default credentials
# Edit these to point at the desired endpoint
cp docker/.docker-env.example docker/.docker-env

# Run the testing container
docker-compose --file docker/docker-compose.yml up --build

# Run the tests
docker exec -it demon pytest src/
```

## External Documentation

The automated functional testing framework relies on the following technologies which developers interface with directly:

* [Docker](https://docs.docker.com/)
* [Python](https://www.python.org/doc/)
* [Pytest](https://docs.pytest.org/en/stable/)
* [Requests](https://requests.readthedocs.io/en/master/)


text:
[pytest]
python_files = *.py


text:
import json
import pytest
import requests
import uuid
import random
import string
from src import env, utils

MAX_SHORTNAME_LENGTH = 32

@pytest.fixture
def org_admin_headers():
    """ create an org and an admin user for that org, return user's headers """
    letters = string.ascii_lowercase
    org = ''.join(random.choice(letters)
                  for i in range(10))  # generate random org name
    org_res = requests.post(
        f'{env.AWG_BASE_URL}/api/org',  # secretariat creates an org
        headers=utils.BASE_HEADERS,
        json={'name': org, 'short_name': org}
    )
    user = str(uuid.uuid4())  # dummy user name
    user_res = requests.post(
        # secretariat creates a user that belongs to the new org
        f'{env.AWG_BASE_URL}/api/org/{org}/user',
        headers=utils.BASE_HEADERS,
        json={'username': user}
    )
    secret = json.loads(user_res.content.decode())[
        'created']['secret']  # dummy user API key
    admin_res = requests.put(
        f'{env.AWG_BASE_URL}/api/org/{org}/user/{user}',
        headers=utils.BASE_HEADERS,
        # secretariat adds the ADMIN role to the new user
        params={'active_roles.add': 'ADMIN'}
    )
    return {
        'CVE-API-KEY': secret,
        'CVE-API-ORG': org,
        'CVE-API-USER': user
    }

@pytest.fixture
def bulk_download_user_headers():
    """ create an org and user for BULK_DOWNLOAD testing, return user's headers """
    letters = string.ascii_lowercase
    org = ''.join(random.choice(letters)
                  for i in range(10))  # generate random org name
    org_res = requests.post(
        f'{env.AWG_BASE_URL}/api/org',  # secretariat creates an org
        headers=utils.BASE_HEADERS,
        json={'name': org, 'short_name': org, 'authority': {'active_roles': ['BULK_DOWNLOAD']}, \
        'policies': {'id_quota': 0}}
    )
    user = str(uuid.uuid4())  # dummy user name
    user_res = requests.post(
        # secretariat creates a user that belongs to the new org
        f'{env.AWG_BASE_URL}/api/org/{org}/user',
        headers=utils.BASE_HEADERS,
        json={'username': user}
    )

    secret = json.loads(user_res.content.decode())[
        'created']['secret']  # dummy user API key
    return {
        'CVE-API-KEY': secret,
        'CVE-API-ORG': org,
        'CVE-API-USER': user
    }

@pytest.fixture
def reg_user_headers():
    """ create an org and an regular user for that org, return user's headers """
    letters = string.ascii_lowercase
    org = ''.join(random.choice(letters)
                  for i in range(10))  # generate random org name
    org_res = requests.post(
        f'{env.AWG_BASE_URL}/api/org',  # secretariat creates an org
        headers=utils.BASE_HEADERS,
        json={'name': org, 'short_name': org}
    )
    user = str(uuid.uuid4())  # dummy user name
    user_res = requests.post(
        # secretariat creates a user that belongs to the new org
        f'{env.AWG_BASE_URL}/api/org/{org}/user',
        headers=utils.BASE_HEADERS,
        json={'username': user}
    )

    secret = json.loads(user_res.content.decode())[
        'created']['secret']  # dummy user API key
    return {
        'CVE-API-KEY': secret,
        'CVE-API-ORG': org,
        'CVE-API-USER': user
    }

# TEST RE-ORDERING
# ==============================================================================
# Tests are re-ordered, and since the default `config/.bashrc` test aliases use
# Pytest's `-x` fail-fast flag, the tests that are run build in complexity and
# earlier failures likely indicate problems that could cause many tests to fail
# (while later failures likely indicate isolated problems).

# The following ordering is currently used:
# 1. `demon` tests which test the test infrastructure and app health checks
# 2. `elements` tests which check the nature of elements on each page; these
# serve as canaries for when a page may have diverged from its corresponding
# tests
# 3. `page_url`, `header`, `cancel` tests are consistent across pages and may
# indicate a broader app change, for instance a new navigation bar; these tests
# also exercise the (above) page fixtures with generic functionality
# 4. `mitre` tests help to ensure, for instance, that the CPS under test has
# a MITRE CNA and its ID quota is >99,998
# 5. all other "systems" tests
# 6. all "interfaces" tests

def pytest_collection_modifyitems(items):
    # arranges tests with these priority elements first
    prioritize = ['demon', 'elements', 'page_url', 'header', 'cancel', 'mitre']
    arranged = 0
    for tag in prioritize:
        for i in range(arranged, len(items)):
            if tag in items[i].name:
                items[arranged], items[i] = items[i], items[arranged]
                arranged += 1

    # move the `interface` tests to the end
    items = list(reversed(items))
    deprioritize = ['interface']
    deranged = 0
    for tag in deprioritize:
        for i in range(deranged, len(items)):
            if tag in items[i].name:
                items[deranged], items[i] = items[i], items[deranged]
                deranged += 1
    items = list(reversed(items))

# PYTEST REPEAT FLAG
# =============================================================================
# give users the option to run any collected tests a flag-set number of times,
# which can help check for non-deterministic failures

def pytest_addoption(parser):
    parser.addoption(
        '--repeat', action='store',
        help='Set the number of times to repeat each test.')

def pytest_generate_tests(metafunc):
    repeat_arg = metafunc.config.option.repeat
    if repeat_arg is not None:
        repeat_fixture = 'repeat_tests'
        n_repeat = int(repeat_arg)
        assert n_repeat > 0

        # test repetition is achieved by adding a fixture to each test, and by
        # using the `pytest.mark.parameterize` pattern to parameterize on the
        # following `repeat_tests` fixture; without changing its value the test
        # just runs `n_repeat` times
        metafunc.fixturenames.append(repeat_fixture)
        metafunc.parametrize(repeat_fixture, range(n_repeat))


text:
import os

AWG_API_KEY = os.environ.get('AWG_API_KEY')
AWG_BASE_URL = os.environ.get('AWG_BASE_URL')
AWG_USER_NAME = os.environ.get('AWG_USER_NAME')

# run performance tests
RUN_PERFORMANCE_TESTS = os.environ.get('RUN_PERFORMANCE_TESTS', 'False')


text:
#!/bin/bash
input="./src/testOutput.txt"
result="Tests passed"

while IFS= read -r line
do
  if grep -q "failed" <<< "$line"; then
    result="Tests failed"
  fi
done < "$input"
echo "$result"


text:
import datetime as dt
import json
from src import env

import logging

logger = logging.getLogger(__name__)

CURRENT_YEAR = dt.datetime.now().year

# HTTP response codes
HTTP_OK = 200

# headers attached to every cve-services request
# org is always mitre, api user depends on setup, but is likely `cps`
BASE_HEADERS = {
    'CVE-API-KEY': env.AWG_API_KEY,
    'CVE-API-ORG': 'mitre',
    'CVE-API-USER': env.AWG_USER_NAME
}

# used to check invalid pagination errors
BAD_PAGE_ERROR_DETAILS = [{
    "msg": "Invalid value",
    "param": "page",
    "location": "query"
}]

def assert_contains(response, has_this, count=1):
    if count == 1:
        assert has_this in response.content.decode()
    else:
        assert response.content.decode().count(has_this) == count

def response_contains(response, has_this):
    assert_contains(response, has_this)

def response_contains_json(response, json_key, is_this, msg=None):
    assert json.loads(response.content.decode())[json_key] == is_this, msg

def ok_response_contains(response, has_this):
    assert response.status_code == HTTP_OK
    assert_contains(response, has_this)

def ok_response_contains_json(response, json_key, is_this):
    assert response.status_code == HTTP_OK
    assert json.loads(response.content.decode())[json_key] == is_this

def get_now_timestamp(fmt = '%Y-%m-%dT%H:%M:%S'):
    """
    Return the a string timestamp for the current date and time.

    The default format is ISO8601 without microseconds.
    """
    return dt.datetime.now(tz=dt.timezone.utc).strftime(fmt)


text:
import multiprocessing
import pytest
import random
import requests
import uuid
from src import env, utils
from src.utils import response_contains_json
from src.test.cve_id_tests.cve_id import get_reserve_cve_ids
from src.test.org_user_tests.org import get_org_id_data, post_new_org

N_JOBS = multiprocessing.cpu_count()
ID_QUOTA_LIMIT = 100000  # FIXME: where do I belong?

#### POST /cve-id ####
def test_sequential_id_reservation_limit():
    """ there's no sequential reservation limit, and its been surprisingly easy
    to reserve 100,000 IDs in the past """
    if env.RUN_PERFORMANCE_TESTS == 'False':
        return
    uid = str(uuid.uuid4())
    post_new_org(uid, uid, ID_QUOTA_LIMIT)
    res = get_reserve_cve_ids(ID_QUOTA_LIMIT, utils.CURRENT_YEAR, uid)
    assert res.status_code == 200

def test_concurrent_sequential_id_reservation_limit():
    """ there's no sequential reservation limit, and its been surprisingly easy
    to reserve 100,000 IDs in the past """
    if env.RUN_PERFORMANCE_TESTS == 'False':
        return

    # create organizations
    uid = str(uuid.uuid4())
    for i in range(0, N_JOBS):
        tmp_uid = f'{uid}-{i}'
        post_new_org(tmp_uid, tmp_uid, ID_QUOTA_LIMIT)

    # asynchronously reserve 100k IDs
    year = random.randint(1999, utils.CURRENT_YEAR+1)
    pool = multiprocessing.Pool(N_JOBS)
    responses = [
        pool.apply_async(
            get_reserve_cve_ids,
            args=(ID_QUOTA_LIMIT, year, f'{uid}-{i}'))
        for i in range(0, N_JOBS)]
    pool.close()
    pool.join()
    for res in responses:
        assert res.get().status_code == 200

@pytest.mark.parametrize(
    "batch_type, amount",
    [('sequential', 1000), ('nonsequential', 100)])
def test_concurrent_id_reservation_scaling(batch_type, amount):
    """ This test will take the following procedure:
    1. Create N_JOBS new Orgs
    2. Set each of their ID quotas to 10*amount
    3. Reserve 10 sets of IDs at a time for each Org
    """
    if env.RUN_PERFORMANCE_TESTS == 'False':
        return

    # create organizations
    uid = str(uuid.uuid4())
    for i in range(0, N_JOBS):
        tmp_uid = f'{uid}-{i}'
        post_new_org(tmp_uid, tmp_uid, 10*amount)

    # asynchronously reserve increasing amounts of IDs
    count = 0
    ids = 0
    have_not_failed = True
    while have_not_failed:
        count += 1
        ids += amount  # tracks number of IDs we've reserved
        try:
            year = random.randint(1999, utils.CURRENT_YEAR+1)
            pool = multiprocessing.Pool(N_JOBS)
            responses = [
                pool.apply_async(
                    get_reserve_cve_ids,
                    args=(amount, year, f'{uid}-{i}', batch_type))
                for i in range(0, N_JOBS)]
            pool.close()
            pool.join()
            for res in responses:
                assert res.get().status_code == 200
            for i in range(0, N_JOBS):
                id_data = get_org_id_data(f'{uid}-{i}')
                response_contains_json(id_data, 'total_reserved', ids)
        except AssertionError:
            # we should have received an assertion error when we hit the new
            # org quota we set, at which point we should see 403 for having
            # explicitly exceeded the quota we set for our new org
            try:
                for res in responses:
                    assert res.get().status_code == 403
                    response_contains_json(res.get(), 'error', 'EXCEEDED_ID_QUOTA')
            except Exception:
                # since this test is exploratory, this line is staying, so I
                # can probe exactly what unexpected behavior occurred
                import pdb
                pdb.set_trace()
        except Exception:
            import pdb
            pdb.set_trace()

def test_interleaved_sequential_non_sequential_scaling():
    """ This test will take the following procedure:
    1. Create N_JOBS new Orgs
    2. Set each of their ID quotas to 100000
    3. Reserve either 1000 sequential or 10 non-sequential (randomly) until
       an Org hits its quota
    """
    if env.RUN_PERFORMANCE_TESTS == 'False':
        return

    # create organizations
    uid = str(uuid.uuid4())
    for i in range(0, N_JOBS):
        tmp_uid = f'{uid}-{i}'
        post_new_org(tmp_uid, tmp_uid, 100000)

    # asynchronously reserve increasing amounts of IDs
    count = 0
    ids = 0
    have_not_failed = True
    while have_not_failed:
        year = random.randint(1999, utils.CURRENT_YEAR+1)
        is_sequential = random.choice([True, False])
        if is_sequential:
            batch_type = 'sequential'
            amount = 1000
        else:
            batch_type = 'nonsequential'
            amount = 10
        count += 1
        ids += amount  # tracks number of IDs we've reserved
        try:
            pool = multiprocessing.Pool(N_JOBS)
            responses = [
                pool.apply_async(
                    get_reserve_cve_ids,
                    args=(amount, year, f'{uid}-{i}', batch_type))
                for i in range(0, N_JOBS)]
            pool.close()
            pool.join()
            for res in responses:
                assert res.get().status_code == 200
            for i in range(0, N_JOBS):
                id_data = get_org_id_data(f'{uid}-{i}')
                response_contains_json(id_data, 'total_reserved', ids)
        except AssertionError:
            # we should have received an assertion error when we hit the new
            # org quota we set, at which point we should see 403 for having
            # explicitly exceeded the quota we set for our new org
            try:
                for res in responses:
                    assert res.get().status_code == 403
                    response_contains_json(res.get(), 'error', 'EXCEEDED_ID_QUOTA')
            except Exception:
                # since this test is exploratory, this line is staying, so I
                # can probe exactly what unexpected behavior occurred
                import pdb
                pdb.set_trace()
        except Exception:
            import pdb
            pdb.set_trace()


text:
import copy
import datetime as dt
import json
import pytest
import requests
import time
import uuid
from src import env, utils
from src.utils import (assert_contains, ok_response_contains,
                       response_contains, response_contains_json,
                       get_now_timestamp)

CVE_ID_URL = '/api/cve-id'
cve_id = 'CVE-1999-0001'

#### GET /cve-id:id ####
def test_get_cve_id(org_admin_headers):
    """ the first ID from 1999 should always exist """
    res = requests.get(
        f'{env.AWG_BASE_URL}{CVE_ID_URL}/{cve_id}',
        headers=org_admin_headers
    )
    ok_response_contains(res, cve_id)

def test_get_cve_id_bad_org_header(org_admin_headers):
    """ unauthenticated users can't get full information about IDs """
    uid = str(uuid.uuid4())
    tmp = copy.deepcopy(org_admin_headers)
    tmp['CVE-API-ORG'] = uid
    tmp['CVE-API-USER'] = uid
    res = requests.get(
        f'{env.AWG_BASE_URL}{CVE_ID_URL}/{cve_id}',
        headers=tmp
    )
    assert res.status_code == 200
    assert 'requested_by' not in res.content.decode()

def test_get_cve_id_id(org_admin_headers):
    """ the id parameter must be a string """
    res = requests.get(
        f'{env.AWG_BASE_URL}{CVE_ID_URL}/{cve_id}',
        headers=org_admin_headers
    )
    assert isinstance (cve_id, str)

"""
Since this test suite unfortunately depends on changes from
other tests, we cannot run this test until making sure it would
not affect the rest of the suite.
"""
# #### PUT /cve-id:id ####
# def test_put_cve_id_id_state(org_admin_headers):
#     """ org admin can update id's state """
#     res = requests.put(
#         f'{env.AWG_BASE_URL}{CVE_ID_URL}/{cve_id}',
#         headers=org_admin_headers,
#         params={'state':'REJECTED'}
#     )
#     assert res.status_code == 200

#### POST /cve-id ####
def test_post_cve_id_update_parameters(org_admin_headers):
    """ org admins can update own information """
    res = requests.post(
        f'{env.AWG_BASE_URL}{CVE_ID_URL}',
        headers=org_admin_headers,
        params={
            'amount': '10',
            'batch_type': 'sequential',
            'cve_year': f'{utils.CURRENT_YEAR}',
            'short_name': org_admin_headers['CVE-API-ORG']
        }
    )
    assert res.status_code == 200

def test_post_cve_id_no_params(org_admin_headers):
    """ batch type is the only optional parameter for reserving ids """
    res = requests.post(
        f'{env.AWG_BASE_URL}{CVE_ID_URL}',
        headers=org_admin_headers
    )
    assert res.status_code == 400
    response_contains(res, 'amount')
    response_contains(res, 'cve_year')
    response_contains(res, 'short_name')

def test_post_cve_id_empty_params(org_admin_headers):
    """ cve services doesn't accept id reservation with blank parameters """
    res = requests.post(
        f'{env.AWG_BASE_URL}{CVE_ID_URL}',
        headers=org_admin_headers,
        params={
            'amount': '',
            'batch_type': '',
            'cve_year': '',
            'short_name': ''
        }
    )
    # NOTE: there isn't a `short_name` error here, why?
    assert res.status_code == 400
    response_contains(res, 'amount')
    response_contains(res, 'cve_year')

def test_post_cve_id_wrong_header(org_admin_headers):
    """ org_admin_headers cannot post for 'mitre' org """
    res = requests.post(
        f'{env.AWG_BASE_URL}{CVE_ID_URL}',
        headers=org_admin_headers,
        params={
            'amount': '10',
            'batch_type': 'sequential',
            'cve_year': f'{utils.CURRENT_YEAR}',
            'short_name': 'mitre'
        }
    )
    # NOTE: this error also occurs when short_name is empty, expected error is 'NO_ORG_SHORTNAME'
    # and when short_name is invalid, expected error is 'ORG_DNE'
    assert res.status_code == 403
    response_contains_json(res, 'error', 'ORG_CANNOT_RESERVE_FOR_OTHER')

def test_post_cve_id_empty_year(org_admin_headers):
    """ cve services rejects empty year """
    res = requests.post(
        f'{env.AWG_BASE_URL}{CVE_ID_URL}',
        headers=org_admin_headers,
        params={
            'amount': '10',
            'batch_type': 'sequential',
            'cve_year': '',
            'short_name': 'mitre'
        }
    )
    assert res.status_code == 400
    response_contains_json(res, 'error', 'BAD_INPUT')

def test_post_cve_id_bad_year(org_admin_headers):
    """ cve services rejects year that isn't a 4 digit number"""
    res = requests.post(
        f'{env.AWG_BASE_URL}{CVE_ID_URL}',
        headers=org_admin_headers,
        params={
            'amount': '10',
            'batch_type': 'sequential',
            'cve_year': '20111',
            'short_name': 'mitre'
        }
    )
    assert res.status_code == 400
    response_contains_json(res, 'error', 'BAD_INPUT')

def test_post_cve_id_empty_amount(org_admin_headers):
    """ cve services rejects empty amount """
    res = requests.post(
        f'{env.AWG_BASE_URL}{CVE_ID_URL}',
        headers=org_admin_headers,
        params={
            'amount': '',
            'batch_type': 'sequential',
            'cve_year': f'{utils.CURRENT_YEAR}',
            'short_name': 'mitre'
        }
    )
    assert res.status_code == 400
    response_contains_json(res, 'error', 'BAD_INPUT')

def test_post_cve_id_invalid_amount(org_admin_headers):
    """ cve services rejects amount less than or equal to 0 """
    res = requests.post(
        f'{env.AWG_BASE_URL}{CVE_ID_URL}',
        headers=org_admin_headers,
        params={
            'amount': '-1',
            'batch_type': 'sequential',
            'cve_year': f'{utils.CURRENT_YEAR}',
            'short_name': org_admin_headers['CVE-API-ORG']
        }
    )
    assert res.status_code == 400
    response_contains_json(res, 'error', 'INVALID_AMOUNT')

def test_post_cve_id_no_batch_type(org_admin_headers):
    """ cve services rejects not having a batch type"""
    res = requests.post(
        f'{env.AWG_BASE_URL}{CVE_ID_URL}',
        headers=org_admin_headers,
        params={
            'amount': '10',
            'batch_type': '',
            'cve_year': f'{utils.CURRENT_YEAR}',
            'short_name': org_admin_headers['CVE-API-ORG']
        }
    )
    assert res.status_code == 400
    response_contains_json(res, 'error', 'NO_BATCH_TYPE')

def test_post_cve_id_invalid_batch_type(org_admin_headers):
    """ cve services rejects batch types that aren't 'sequential' or 'nonsequential' """
    res = requests.post(
        f'{env.AWG_BASE_URL}{CVE_ID_URL}',
        headers=org_admin_headers,
        params={
            'amount': '10',
            'batch_type': '---',
            'cve_year': f'{utils.CURRENT_YEAR}',
            'short_name': org_admin_headers['CVE-API-ORG']
        }
    )
    assert res.status_code == 400
    response_contains_json(res, 'error', 'INVALID_BATCH_TYPE')

def test_post_cve_id_bad_amount(org_admin_headers):
    """ api rejects non-numeric amount when requesting IDs """
    res = get_reserve_cve_ids('a', utils.CURRENT_YEAR, org_admin_headers['CVE-API-ORG'])
    assert res.status_code == 400
    assert res.reason == 'Bad Request'
    response_contains_json(res, 'error', 'BAD_INPUT')
    assert_contains(res, 'amount')

def test_post_cve_id_reserve_priority(org_admin_headers):
    """ priority ids can be reserved on behalf of the admin's org """
    res = requests.post(
        f'{env.AWG_BASE_URL}{CVE_ID_URL}',
        headers=org_admin_headers,
        params={
            'amount': '1',
            'cve_year': f'{utils.CURRENT_YEAR}',
            'short_name': org_admin_headers['CVE-API-ORG']
        }
    )
    ok_response_contains(res, f'CVE-{utils.CURRENT_YEAR}-')
    assert json.loads(res.content.decode())['cve_ids']
    assert len(json.loads(res.content.decode())['cve_ids']) == 1

    priority_id = json.loads(res.content.decode())['cve_ids'][0]['cve_id']
    assert int(priority_id.split('-')[-1]) < 20000
    for key in json.loads(res.content.decode())['meta'].keys(): # Check that remaining_quota is in response
        assert key == 'remaining_quota'

# Does this constitute a performance test? The distinction I'm making is that
# performance tests stress the system overall, while these tests try to
# test both "reservation works" and "there's some reasonable amount that we
# can request that doesn't stress the system"
@pytest.mark.parametrize(
    "batch_type, amount",
    [('sequential', 10), ('sequential', 1000),
     ('nonsequential', 1), ('nonsequential', 10)])
def test_post_cve_id_reservation(batch_type, amount, org_admin_headers):
    """ sequential ids can be reserved on behalf of the mitre org """
    res = get_reserve_cve_ids(amount, utils.CURRENT_YEAR, org_admin_headers['CVE-API-ORG'], batch_type)
    ok_response_contains(res, f'CVE-{utils.CURRENT_YEAR}-')
    assert json.loads(res.content.decode())['cve_ids']
    assert len(json.loads(res.content.decode())['cve_ids']) == amount
    # cna and user must exist
    assert 'cna' in res.content.decode()
    assert 'user' in res.content.decode()
    for key in json.loads(res.content.decode())['meta'].keys(): # Check that remaining_quota is in response
        assert key == 'remaining_quota'

def test_post_cve_id_reserve_sequential_over_quota(org_admin_headers):
    """ the services api enforces a max quota of 100,000 """
    res = get_reserve_cve_ids(100001, utils.CURRENT_YEAR, org_admin_headers['CVE-API-ORG'])
    assert res.status_code == 403
    response_contains_json(res, 'error', 'EXCEEDED_ID_QUOTA')

def test_post_cve_id_reserve_nonsequential_over_limit(org_admin_headers):
    """ the services api enforces a max non-sequential limit of 10 """
    res = get_reserve_cve_ids(11, utils.CURRENT_YEAR, org_admin_headers['CVE-API-ORG'], 'nonsequential')
    assert res.status_code == 403
    response_contains_json(res, 'error', 'OVER_NONSEQUENTIAL_MAX_AMOUNT')

#### GET /cve-id ####
def test_get_cve_id_by_time_reserved(org_admin_headers):
    """ we can get ids immediately after reserving them using the time they're
    reserved (noting that this may not work against a shared integration
    environment, we check that at least this many have been reserved) """
    n_ids = 10
    time.sleep(1)
    t_before = get_now_timestamp()
    time.sleep(1)
    res_ids = get_reserve_cve_ids(n_ids, utils.CURRENT_YEAR, org_admin_headers['CVE-API-ORG'])
    time.sleep(1)
    t_after = get_now_timestamp()

    res_get_ids = requests.get(
        f'{env.AWG_BASE_URL}{CVE_ID_URL}',
        headers=utils.BASE_HEADERS,
        params={
            'time_reserved.lt': t_after,
            'time_reserved.gt': t_before
        }
    )
    ok_response_contains(res_get_ids, f'CVE-{utils.CURRENT_YEAR}-')
    assert len(json.loads(res_get_ids.content.decode())['cve_ids']) == n_ids

def test_get_cve_id_by_time_modified(org_admin_headers):
    """ we can get ids immediately after reserving them using the time they're
    reserved (noting that this may not work against a shared integration
    environment, we check that at least this many have been reserved) """
    n_ids = 10
    time.sleep(1)
    t_before = get_now_timestamp()
    t_before_alt = dt.datetime.now(tz=dt.timezone.utc).strftime('%Y-%m-%dT%H:%M:%S+00:00')
    time.sleep(1)
    res_ids = get_reserve_cve_ids(n_ids, utils.CURRENT_YEAR, org_admin_headers['CVE-API-ORG'])
    time.sleep(1)
    t_after = get_now_timestamp()
    t_after_alt = dt.datetime.now(tz=dt.timezone.utc).strftime('%Y-%m-%dT%H:%M:%S+00:00')

    res_get_ids = requests.get(
        f'{env.AWG_BASE_URL}{CVE_ID_URL}',
        headers=utils.BASE_HEADERS,
        params={
            'time_modified.lt': t_after,
            'time_modified.gt': t_before
        }
    )
    # Test alternate timezone format
    res_get_ids_alt = requests.get(
        f'{env.AWG_BASE_URL}{CVE_ID_URL}',
        headers=utils.BASE_HEADERS,
        params={
            'time_modified.lt': t_after_alt,
            'time_modified.gt': t_before_alt
        }
    )
    ok_response_contains(res_get_ids, f'CVE-{utils.CURRENT_YEAR}-')
    assert len(json.loads(res_get_ids.content.decode())['cve_ids']) == n_ids
    ok_response_contains(res_get_ids_alt, f'CVE-{utils.CURRENT_YEAR}-')
    assert len(json.loads(res_get_ids_alt.content.decode())['cve_ids']) == n_ids

def test_get_cve_id_with_params(org_admin_headers):
    """ org admin can retrieve ids"""
    res = requests.get(
        f'{env.AWG_BASE_URL}{CVE_ID_URL}',
        headers=org_admin_headers,
        params={
            'page': 1,
            'state': 'PUBLISHED',
            'cve_id_year': 2011
        }
    )
    assert res.status_code == 200

def test_get_cve_id_empty_parameters(org_admin_headers):
    """ cannot get id with empty parameters """
    res = requests.get(
        f'{env.AWG_BASE_URL}{CVE_ID_URL}',
        headers=org_admin_headers,
        params={
            'page': ' ',
            'state': ' ',
            'cve_id_year': ' ',
            'time_reserved.lt': ' ',
            'time_reserved.gt': ' ',
            'time_modified.lt': ' ',
            'time_modified.gt': ' '
        }
    )
    assert res.status_code == 400
    response_contains(res, 'page')
    response_contains(res, 'state')
    response_contains(res, 'cve_id_year')
    response_contains(res, 'time_reserved.lt')
    response_contains(res, 'time_reserved.gt')
    response_contains(res, 'time_modified.lt')
    response_contains(res, 'time_modified.gt')
    response_contains_json(res, 'error', 'BAD_INPUT')

def test_get_cve_id_page_format_number(org_admin_headers):
    """ page must be an integer' """
    res = requests.get(
        f'{env.AWG_BASE_URL}{CVE_ID_URL}',
        headers=org_admin_headers,
        params={
            'page': 'test',
        }
    )
    assert res.status_code == 400
    response_contains_json(res, 'error', 'BAD_INPUT')

def test_get_cve_id_page_limit(org_admin_headers):
    """ page must be greater than or equal to 1' """
    res = requests.get(
        f'{env.AWG_BASE_URL}{CVE_ID_URL}',
        headers=org_admin_headers,
        params={
            'page': '-1',
        }
    )
    assert res.status_code == 400
    response_contains_json(res, 'error', 'BAD_INPUT')

def test_get_cve_id_state_in_choices(org_admin_headers):
    """ state parameter can only be 'REJECTED', 'PUBLISHED' or 'RESERVED' """
    res = requests.get(
        f'{env.AWG_BASE_URL}{CVE_ID_URL}',
        headers=org_admin_headers,
        params={
            'state': 'TEST',
        }
    )
    assert res.status_code == 400
    response_contains_json(res, 'error', 'BAD_INPUT')

def test_get_cve_id_year_format_with_letters(org_admin_headers):
    """ cve_id_year format cannot have letters  """
    res = requests.get(
        f'{env.AWG_BASE_URL}{CVE_ID_URL}',
        headers=org_admin_headers,
        params={
            'cve_id_year': 'test',
        }
    )
    assert res.status_code == 400
    response_contains_json(res, 'error', 'BAD_INPUT')

def test_get_cve_id_year_format_with_digits(org_admin_headers):
    """ cve_id_year format must be 4 digits only  """
    res = requests.get(
        f'{env.AWG_BASE_URL}{CVE_ID_URL}',
        headers=org_admin_headers,
        params={
            'cve_id_year': '20111',
        }
    )
    assert res.status_code == 400
    response_contains_json(res, 'error', 'BAD_INPUT')

def test_get_cve_id_available_state(org_admin_headers):
    """ CVE ID filter endpoint does not return any IDs with state 'AVAILABLE' """
    res = requests.get(
        f'{env.AWG_BASE_URL}{CVE_ID_URL}',
        headers=org_admin_headers,
        params={
            'page': 1,
            'state': 'PUBLISHED',
            'cve_id_year': 2011
        }
    )
    assert res.status_code == 200
    assert 'AVAILABLE' not in res.content.decode()

# CVE-ID ENDPOINT UTILITIES
# ==============================================================================
# these are unique to the `{CVE_ID_URL}` endpoint for the AWG system

def get_reserve_cve_ids(
        amount, year, cna_short_name, batch_type='sequential'):
    return requests.post(
        f'{env.AWG_BASE_URL}{CVE_ID_URL}',
        headers=utils.BASE_HEADERS,
        params={
            'amount': f'{amount}',
            'batch_type': batch_type,
            'cve_year': f'{year}',
            'short_name': cna_short_name
        }
    )


text:
import pytest
import requests
from src import env, utils
from src.utils import response_contains, response_contains_json

CVE_ID_RANGE_URL = '/api/cve-id-range'

#### POST /cve-id-range/:year ####
@pytest.mark.parametrize(
    "choose_year",
    [str(x) for x in range(1999, utils.CURRENT_YEAR + 1)])
def test_post_cve_id_range_already_exists(choose_year):
    """ ranges for 1999 to the current year must exist """
    res = requests.post(
        f'{env.AWG_BASE_URL}{CVE_ID_RANGE_URL}/{choose_year}',
        headers=utils.BASE_HEADERS
    )
    assert res.status_code == 400
    response_contains(
        res,
        (f'document for year {choose_year} was not created '
         'because it already exists.'))
    response_contains_json(res, 'error', 'YEAR_RANGE_EXISTS')


text:
import copy
import datetime as dt
import json
import pytest
import requests
import time
import uuid
from src import env, utils
from src.utils import (assert_contains, ok_response_contains,
                       response_contains, response_contains_json,
                       get_now_timestamp)

CVE_ID_URL = '/api/cve-id'
cve_id = 'CVE-1999-0001'

#### GET /cve-id/:id ####
def test_get_cve_id(reg_user_headers):
    """ the first ID from 1999 should always exist """
    res = requests.get(
        f'{env.AWG_BASE_URL}{CVE_ID_URL}/{cve_id}',
        headers=reg_user_headers
    )
    ok_response_contains(res, cve_id)

def test_get_cve_id_bad_org_header(reg_user_headers):
    """ unauthenticated users can't get full information about IDs """
    uid = str(uuid.uuid4())
    tmp = copy.deepcopy(reg_user_headers)
    tmp['CVE-API-ORG'] = uid
    tmp['CVE-API-USER'] = uid
    res = requests.get(
        f'{env.AWG_BASE_URL}{CVE_ID_URL}/{cve_id}',
        headers=tmp
    )
    assert res.status_code == 200
    assert 'requested_by' not in res.content.decode()

def test_get_cve_id_id(reg_user_headers):
    """ the id parameter must be a string """
    res = requests.get(
        f'{env.AWG_BASE_URL}{CVE_ID_URL}/{cve_id}',
        headers=reg_user_headers
    )
    assert isinstance (cve_id, str)

"""
Regular users now CAN update CVE ID state.
This test should be reinstituted once the problem
of this test suite using the same CVE ID for all
these tests is resolved. Tests should be independent
of one another but these tests are highly dependent
on each other's state changes.
"""
#### PUT /cve-id/:id ####
# def test_put_cve_id_id_state(reg_user_headers):
#     """ regular user cannot update id's state """
#     res = requests.put(
#         f'{env.AWG_BASE_URL}{CVE_ID_URL}/{cve_id}',
#         headers=reg_user_headers,
#         params={'state':'RESERVED'}
#     )
#     assert res.status_code == 200
#     # This should not result in an error, but a success message.
#     response_contains_json(res, 'error', 'SECRETARIAT_ONLY')

#### POST /cve-id ####
def test_post_cve_id_update_parameters(reg_user_headers):
    """ org users can update own information """
    res = requests.post(
        f'{env.AWG_BASE_URL}{CVE_ID_URL}',
        headers=reg_user_headers,
        params={
            'amount': '10',
            'batch_type': 'sequential',
            'cve_year': f'{utils.CURRENT_YEAR}',
            'short_name': reg_user_headers['CVE-API-ORG']
        }
    )
    assert res.status_code == 200

def test_post_cve_id_no_params(reg_user_headers):
    """ batch type is the only optional parameter for reserving ids """
    res = requests.post(
        f'{env.AWG_BASE_URL}{CVE_ID_URL}',
        headers=reg_user_headers
    )
    assert res.status_code == 400
    response_contains(res, 'amount')
    response_contains(res, 'cve_year')
    response_contains(res, 'short_name')

def test_post_cve_id_empty_params(reg_user_headers):
    """ cve services doesn't accept id reservation with blank parameters """
    res = requests.post(
        f'{env.AWG_BASE_URL}{CVE_ID_URL}',
        headers=reg_user_headers,
        params={
            'amount': '',
            'batch_type': '',
            'cve_year': '',
            'short_name': ''
        }
    )
    # NOTE: there isn't a `short_name` error here, why?
    assert res.status_code == 400
    response_contains(res, 'amount')
    response_contains(res, 'cve_year')

def test_post_cve_id_wrong_header(reg_user_headers):
    """ reg_user_headers cannot post for 'mitre' org """
    res = requests.post(
        f'{env.AWG_BASE_URL}{CVE_ID_URL}',
        headers=reg_user_headers,
        params={
            'amount': '10',
            'batch_type': 'sequential',
            'cve_year': f'{utils.CURRENT_YEAR}',
            'short_name': 'mitre'
        }
    )
    # NOTE: this error also occurs when short_name is empty and invalid,
    # expected error is 'NO_ORG_SHORTNAME' and 'ORG_DNE' respectively
    assert res.status_code == 403
    response_contains_json(res, 'error', 'ORG_CANNOT_RESERVE_FOR_OTHER')

def test_post_cve_id_empty_year(reg_user_headers):
    """ cve services rejects empty year """
    res = requests.post(
        f'{env.AWG_BASE_URL}{CVE_ID_URL}',
        headers=reg_user_headers,
        params={
            'amount': '10',
            'batch_type': 'sequential',
            'cve_year': '',
            'short_name': 'mitre'
        }
    )
    assert res.status_code == 400
    response_contains_json(res, 'error', 'BAD_INPUT')

def test_get_cve_id_bulk_download_user(bulk_download_user_headers):
    """ bulk download user should not be a CNA so can't reserve ids """
    res = requests.post(
        f'{env.AWG_BASE_URL}{CVE_ID_URL}',
        headers=bulk_download_user_headers,
        params={
            'amount': '1',
            'cve_year': f'{utils.CURRENT_YEAR}',
            'short_name': bulk_download_user_headers['CVE-API-ORG']
        }
    )
    assert res.status_code == 403
    response_contains_json(res, 'error', 'CNA_ONLY')

def test_post_cve_id_bad_year(reg_user_headers):
    """ cve services rejects year that isn't a 4 digit number"""
    res = requests.post(
        f'{env.AWG_BASE_URL}{CVE_ID_URL}',
        headers=reg_user_headers,
        params={
            'amount': '10',
            'batch_type': 'sequential',
            'cve_year': '20111',
            'short_name': 'mitre'
        }
    )
    assert res.status_code == 400
    response_contains_json(res, 'error', 'BAD_INPUT')

def test_post_cve_id_empty_amount(reg_user_headers):
    """ cve services rejects empty amount """
    res = requests.post(
        f'{env.AWG_BASE_URL}{CVE_ID_URL}',
        headers=reg_user_headers,
        params={
            'amount': '',
            'batch_type': 'sequential',
            'cve_year': f'{utils.CURRENT_YEAR}',
            'short_name': 'mitre'
        }
    )
    assert res.status_code == 400
    response_contains_json(res, 'error', 'BAD_INPUT')

def test_post_cve_id_invalid_amount(reg_user_headers):
    """ cve services rejects amount less than or equal to 0 """
    res = requests.post(
        f'{env.AWG_BASE_URL}{CVE_ID_URL}',
        headers=reg_user_headers,
        params={
            'amount': '-1',
            'batch_type': 'sequential',
            'cve_year': f'{utils.CURRENT_YEAR}',
            'short_name': reg_user_headers['CVE-API-ORG']
        }
    )
    assert res.status_code == 400
    response_contains_json(res, 'error', 'INVALID_AMOUNT')

def test_post_cve_id_no_batch_type(reg_user_headers):
    """ cve services rejects not having a batch type"""
    res = requests.post(
        f'{env.AWG_BASE_URL}{CVE_ID_URL}',
        headers=reg_user_headers,
        params={
            'amount': '10',
            'batch_type': '',
            'cve_year': f'{utils.CURRENT_YEAR}',
            'short_name': reg_user_headers['CVE-API-ORG']
        }
    )
    assert res.status_code == 400
    response_contains_json(res, 'error', 'NO_BATCH_TYPE')

def test_post_cve_id_invalid_batch_type(reg_user_headers):
    """ cve services rejects batch types that aren't 'sequential' or 'nonsequential' """
    res = requests.post(
        f'{env.AWG_BASE_URL}{CVE_ID_URL}',
        headers=reg_user_headers,
        params={
            'amount': '10',
            'batch_type': '---',
            'cve_year': f'{utils.CURRENT_YEAR}',
            'short_name':  reg_user_headers['CVE-API-ORG']
        }
    )
    assert res.status_code == 400
    response_contains_json(res, 'error', 'INVALID_BATCH_TYPE')

def test_post_cve_id_bad_amount(reg_user_headers):
    """ api rejects non-numeric amount when requesting IDs """
    res = get_reserve_cve_ids('a', utils.CURRENT_YEAR, reg_user_headers['CVE-API-USER'])
    assert res.status_code == 400
    assert res.reason == 'Bad Request'
    response_contains_json(res, 'error', 'BAD_INPUT')
    assert_contains(res, 'amount')

def test_post_cve_id_reserve_priority(reg_user_headers):
    """ priority ids can be reserved on behalf of the general user org """
    res = requests.post(
        f'{env.AWG_BASE_URL}{CVE_ID_URL}',
        headers=reg_user_headers,
        params={
            'amount': '1',
            'cve_year': f'{utils.CURRENT_YEAR}',
            'short_name': reg_user_headers['CVE-API-ORG']
        }
    )
    ok_response_contains(res, f'CVE-{utils.CURRENT_YEAR}-')
    assert json.loads(res.content.decode())['cve_ids']
    assert len(json.loads(res.content.decode())['cve_ids']) == 1

    priority_id = json.loads(res.content.decode())['cve_ids'][0]['cve_id']
    assert int(priority_id.split('-')[-1]) < 20000
    for key in json.loads(res.content.decode())['meta'].keys(): # Check that remaining_quota is in response
        assert key == 'remaining_quota'

# Does this constitute a performance test? The distinction I'm making is that
# performance tests stress the system overall, while these tests try to
# test both "reservation works" and "there's some reasonable amount that we
# can request that doesn't stress the system"
@pytest.mark.parametrize(
    "batch_type, amount",
    [('sequential', 10), ('sequential', 1000),
     ('nonsequential', 1), ('nonsequential', 10)])
def test_post_cve_id_reservation(batch_type, amount, reg_user_headers):
    """ sequential ids can be reserved on behalf of the mitre org """
    res = get_reserve_cve_ids(amount, utils.CURRENT_YEAR, reg_user_headers['CVE-API-ORG'], batch_type)
    ok_response_contains(res, f'CVE-{utils.CURRENT_YEAR}-')
    assert json.loads(res.content.decode())['cve_ids']
    assert len(json.loads(res.content.decode())['cve_ids']) == amount
    # cna and user must exist
    assert 'cna' in res.content.decode()
    assert 'user' in res.content.decode()
    for key in json.loads(res.content.decode())['meta'].keys(): # Check that remaining_quota is in response
        assert key == 'remaining_quota'

def test_post_cve_id_reserve_sequential_over_quota(reg_user_headers):
    """ the services api enforces a max quota of 100,000 """
    res = get_reserve_cve_ids(100001, utils.CURRENT_YEAR, reg_user_headers['CVE-API-ORG'])
    assert res.status_code == 403
    response_contains_json(res, 'error', 'EXCEEDED_ID_QUOTA')

def test_post_cve_id_reserve_nonsequential_over_limit(reg_user_headers):
    """ the services api enforces a max non-sequential limit of 10 """
    res = get_reserve_cve_ids(11, utils.CURRENT_YEAR, reg_user_headers['CVE-API-ORG'], 'nonsequential')
    assert res.status_code == 403
    response_contains_json(res, 'error', 'OVER_NONSEQUENTIAL_MAX_AMOUNT')

#### GET /cve-id ####
def test_get_cve_id_by_time_reserved(reg_user_headers):
    """ we can get ids immediately after reserving them using the time they're
    reserved (noting that this may not work against a shared integration
    environment, we check that at least this many have been reserved) """
    n_ids = 10
    time.sleep(1)
    t_before = get_now_timestamp()
    time.sleep(1)
    res_ids = get_reserve_cve_ids(n_ids, utils.CURRENT_YEAR, reg_user_headers['CVE-API-ORG'])
    time.sleep(1)
    t_after = get_now_timestamp()

    res_get_ids = requests.get(
        f'{env.AWG_BASE_URL}{CVE_ID_URL}',
        headers=utils.BASE_HEADERS,
        params={
            'time_reserved.lt': t_after,
            'time_reserved.gt': t_before
        }
    )
    ok_response_contains(res_get_ids, f'CVE-{utils.CURRENT_YEAR}-')
    assert len(json.loads(res_get_ids.content.decode())['cve_ids']) == n_ids

def test_get_cve_id_by_time_modified(reg_user_headers):
    """ we can get ids immediately after reserving them using the time they're
    reserved (noting that this may not work against a shared integration
    environment, we check that at least this many have been reserved) """
    n_ids = 10
    time.sleep(1)
    t_before = get_now_timestamp()
    time.sleep(1)
    res_ids = get_reserve_cve_ids(n_ids, utils.CURRENT_YEAR, reg_user_headers['CVE-API-ORG'])
    time.sleep(1)
    t_after = get_now_timestamp()

    res_get_ids = requests.get(
        f'{env.AWG_BASE_URL}{CVE_ID_URL}',
        headers=utils.BASE_HEADERS,
        params={
            'time_modified.lt': t_after,
            'time_modified.gt': t_before
        }
    )
    ok_response_contains(res_get_ids, f'CVE-{utils.CURRENT_YEAR}-')
    assert len(json.loads(res_get_ids.content.decode())['cve_ids']) == n_ids

def test_get_cve_id_with_params(reg_user_headers):
    """ org user can retrieve ids"""
    res = requests.get(
        f'{env.AWG_BASE_URL}{CVE_ID_URL}',
        headers=reg_user_headers,
        params={
            'page': 1,
            'state': 'PUBLISHED',
            'cve_id_year': 2011
        }
    )
    assert res.status_code == 200

def test_get_cve_id_empty_parameters(reg_user_headers):
    """ cannot get id with empty parameters """
    res = requests.get(
        f'{env.AWG_BASE_URL}{CVE_ID_URL}',
        headers=reg_user_headers,
        params={
            'page': ' ',
            'state': ' ',
            'cve_id_year': ' ',
            'time_reserved.lt': ' ',
            'time_reserved.gt': ' ',
            'time_modified.lt': ' ',
            'time_modified.gt': ' '
        }
    )
    assert res.status_code == 400
    response_contains(res, 'page')
    response_contains(res, 'state')
    response_contains(res, 'cve_id_year')
    response_contains(res, 'time_reserved.lt')
    response_contains(res, 'time_reserved.gt')
    response_contains(res, 'time_modified.lt')
    response_contains(res, 'time_modified.gt')
    response_contains_json(res, 'error', 'BAD_INPUT')

def test_get_cve_id_page_format_number(reg_user_headers):
    """ page must be an integer' """
    res = requests.get(
        f'{env.AWG_BASE_URL}{CVE_ID_URL}',
        headers=reg_user_headers,
        params={
            'page': 'test',
        }
    )
    assert res.status_code == 400
    response_contains_json(res, 'error', 'BAD_INPUT')

def test_get_cve_id_page_limit(reg_user_headers):
    """ page must be greater than or equal to 1' """
    res = requests.get(
        f'{env.AWG_BASE_URL}{CVE_ID_URL}',
        headers=reg_user_headers,
        params={
            'page': '-1',
        }
    )
    assert res.status_code == 400
    response_contains_json(res, 'error', 'BAD_INPUT')

def test_get_cve_id_state_in_choices(reg_user_headers):
    """ state parameter can only be 'REJECTED', 'PUBLISHED' or 'RESERVED' """
    res = requests.get(
        f'{env.AWG_BASE_URL}{CVE_ID_URL}',
        headers=reg_user_headers,
        params={
            'state': 'TEST',
        }
    )
    assert res.status_code == 400
    response_contains_json(res, 'error', 'BAD_INPUT')

def test_get_cve_id_year_format_with_letters(reg_user_headers):
    """ cve_id_year format cannot have letters  """
    res = requests.get(
        f'{env.AWG_BASE_URL}{CVE_ID_URL}',
        headers=reg_user_headers,
        params={
            'cve_id_year': 'test',
        }
    )
    assert res.status_code == 400
    response_contains_json(res, 'error', 'BAD_INPUT')

def test_get_cve_id_year_format_with_digits(reg_user_headers):
    """ cve_id_year format must be 4 digits only  """
    res = requests.get(
        f'{env.AWG_BASE_URL}{CVE_ID_URL}',
        headers=reg_user_headers,
        params={
            'cve_id_year': '20111',
        }
    )
    assert res.status_code == 400
    response_contains_json(res, 'error', 'BAD_INPUT')

def test_get_cve_id_available_state(reg_user_headers):
    """ CVE ID filter endpoint does not return any IDs with state 'AVAILABLE' """
    res = requests.get(
        f'{env.AWG_BASE_URL}{CVE_ID_URL}',
        headers=reg_user_headers,
        params={
            'page': 1,
            'state': 'PUBLISHED',
            'cve_id_year': 2011
        }
    )
    assert res.status_code == 200
    assert 'AVAILABLE' not in res.content.decode()

# CVE-ID ENDPOINT UTILITIES
# ==============================================================================
# these are unique to the `{CVE_ID_URL}` endpoint for the AWG system

def get_reserve_cve_ids(
        amount, year, cna_short_name, batch_type='sequential'):
    return requests.post(
        f'{env.AWG_BASE_URL}{CVE_ID_URL}',
        headers=utils.BASE_HEADERS,
        params={
            'amount': f'{amount}',
            'batch_type': batch_type,
            'cve_year': f'{year}',
            'short_name': cna_short_name
        }
    )


text:
import copy
import datetime as dt
import json
import pytest
import requests
import time
import uuid
from src import env, utils
from src.utils import (assert_contains, ok_response_contains,
                       response_contains, response_contains_json,
                       get_now_timestamp)

CVE_ID_URL = '/api/cve-id'
cve_id = 'CVE-1999-0002'

#### GET /cve-id/:id ####

def test_get_cve_id():
    """ the first ID from 1999 should always exist """
    res = requests.get(
        f'{env.AWG_BASE_URL}{CVE_ID_URL}/{cve_id}',
        headers=utils.BASE_HEADERS
    )
    ok_response_contains(res, cve_id)

def test_get_cve_id_bad_org_header():
    """ unauthenticated users can't get full information about IDs """
    uid = str(uuid.uuid4())
    tmp = copy.deepcopy(utils.BASE_HEADERS)
    tmp['CVE-API-ORG'] = uid
    tmp['CVE-API-USER'] = uid
    res = requests.get(
        f'{env.AWG_BASE_URL}{CVE_ID_URL}/{cve_id}',
        headers=tmp
    )
    assert res.status_code == 200
    assert 'requested_by' not in res.content.decode()

def test_get_cve_id_id():
    """ the id parameter must be a string """
    res = requests.get(
        f'{env.AWG_BASE_URL}{CVE_ID_URL}/{cve_id}',
        headers=utils.BASE_HEADERS
    )
    assert isinstance(cve_id, str)

#### PUT /cve-id/:id ####
def test_put_cve_id_id():
    """ an id can be updated to rejected and then back to reserved """
    res = requests.put(
        f'{env.AWG_BASE_URL}{CVE_ID_URL}/{cve_id}',
        headers=utils.BASE_HEADERS,
        params={'state': 'REJECTED'}
    )
    ok_response_contains(res, f'{cve_id} was successfully updated.')

    res = requests.put(
        f'{env.AWG_BASE_URL}{CVE_ID_URL}/{cve_id}',
        headers=utils.BASE_HEADERS,
        params={'state': 'RESERVED'}
    )
    ok_response_contains(res, f'{cve_id} was successfully updated.')

"""
This test will be reinstated when the code is completed to prevent CVE IDs
from being changed to the PUBLISHED state.
"""
# def test_put_cve_id_id_state_published():
#     """ an id's state cannot be set to published """
#     res = requests.put(
#         f'{env.AWG_BASE_URL}{CVE_ID_URL}/{cve_id}',
#         headers=utils.BASE_HEADERS,
#         params={'state':'REJECTED'}
#     )
#     ok_response_contains(res, f'{cve_id} was successfully updated.')

#     res = requests.put(
#         f'{env.AWG_BASE_URL}{CVE_ID_URL}/{cve_id}',
#         headers=utils.BASE_HEADERS,
#         params={'state':'PUBLISHED'}
#     )
#     assert res.status_code == 400
#     response_contains_json(
#         res, 'message',
#         'Cannot change the state to PUBLISHED.')

def test_put_cve_id_id_state_blarg():
    """ an id's state cannot be changed to a non-existent state """
    res = requests.put(
        f'{env.AWG_BASE_URL}{CVE_ID_URL}/{cve_id}',
        headers=utils.BASE_HEADERS,
        params={'state': 'BLARG'}
    )
    assert res.status_code == 400
    assert 'BLARG' not in res.content.decode()
    response_contains(res, 'state')
    response_contains_json(res, 'message', 'Parameters were invalid')

def test_put_cve_id_id_no_params():
    """ cve id endpoint ... accepts an empty query set? """
    res = requests.put(
        f'{env.AWG_BASE_URL}{CVE_ID_URL}/{cve_id}',
        headers=utils.BASE_HEADERS
    )
    # NOTE: should this behave as if an update was successful, when no params
    # were included and nothing was actually updated?
    ok_response_contains(res, f'{cve_id} was successfully updated.')

def test_put_cve_id_id_empty_params():
    """ cve services id update endpoint fails for empty query parameters """
    res = requests.put(
        f'{env.AWG_BASE_URL}{CVE_ID_URL}/{cve_id}',
        headers=utils.BASE_HEADERS,
        params={'state': '', 'cna': ''}
    )
    assert res.status_code == 400
    response_contains(res, 'state')
    response_contains_json(res, 'message', 'Parameters were invalid')

#### POST /cve-id ####

def test_post_cve_id_no_params():
    """ batch type is the only optional parameter for reserving ids """
    res = requests.post(
        f'{env.AWG_BASE_URL}{CVE_ID_URL}',
        headers=utils.BASE_HEADERS
    )
    assert res.status_code == 400
    response_contains(res, 'amount')
    response_contains(res, 'cve_year')
    response_contains(res, 'short_name')

def test_post_cve_id_empty_params():
    """ cve services doesn't accept id reservation with blank parameters """
    res = requests.post(
        f'{env.AWG_BASE_URL}{CVE_ID_URL}',
        headers=utils.BASE_HEADERS,
        params={
            'amount': '',
            'batch_type': '',
            'cve_year': '',
            'short_name': ''
        }
    )
    # NOTE: there isn't a `short_name` error here, why?
    assert res.status_code == 400
    response_contains(res, 'amount')
    response_contains(res, 'cve_year')

def test_post_cve_id_empty_short_name():
    """ cve services rejects empty short name separately """
    res = requests.post(
        f'{env.AWG_BASE_URL}{CVE_ID_URL}',
        headers=utils.BASE_HEADERS,
        params={
            'amount': '10',
            'batch_type': 'sequential',
            'cve_year': f'{utils.CURRENT_YEAR}',
            'short_name': ''
        }
    )
    assert res.status_code == 400
    response_contains_json(res, 'error', 'BAD_INPUT')

def test_post_cve_id_organization_access():
    """ services rejects ID reservation for nonexistent Org as identified by short name """
    res = requests.post(
        f'{env.AWG_BASE_URL}{CVE_ID_URL}',
        headers=utils.BASE_HEADERS,
        params={
            'amount': '10',
            'batch_type': 'sequential',
            'cve_year': f'{utils.CURRENT_YEAR}',
            'short_name': 'test'
        }
    )
    assert res.status_code == 403
    response_contains_json(res, 'error', 'ORG_DNE')

def test_post_cve_id_shortname_format():
    """ cve services rejects shortname that's not a string """
    res = requests.post(
        f'{env.AWG_BASE_URL}{CVE_ID_URL}',
        headers=utils.BASE_HEADERS,
        params={
            'amount': '10',
            'batch_type': 'sequential',
            'cve_year': f'{utils.CURRENT_YEAR}',
            'short_name': '1111'
        }
    )
    assert res.status_code == 403
    response_contains_json(res, 'error', 'ORG_DNE')

def test_post_cve_id_empty_year():
    """ cve services rejects empty year """
    res = requests.post(
        f'{env.AWG_BASE_URL}{CVE_ID_URL}',
        headers=utils.BASE_HEADERS,
        params={
            'amount': '10',
            'batch_type': 'sequential',
            'cve_year': '',
            'short_name': 'mitre'
        }
    )
    assert res.status_code == 400
    response_contains_json(res, 'error', 'BAD_INPUT')

def test_post_cve_id_bad_year():
    """ cve services rejects year that isn't a 4 digit number"""
    res = requests.post(
        f'{env.AWG_BASE_URL}{CVE_ID_URL}',
        headers=utils.BASE_HEADERS,
        params={
            'amount': '10',
            'batch_type': 'sequential',
            'cve_year': '20111',
            'short_name': 'mitre'
        }
    )
    assert res.status_code == 400
    response_contains_json(res, 'error', 'BAD_INPUT')

def test_post_cve_id_empty_amount():
    """ cve services rejects empty amount """
    res = requests.post(
        f'{env.AWG_BASE_URL}{CVE_ID_URL}',
        headers=utils.BASE_HEADERS,
        params={
            'amount': '',
            'batch_type': 'sequential',
            'cve_year': f'{utils.CURRENT_YEAR}',
            'short_name': 'mitre'
        }
    )
    assert res.status_code == 400
    response_contains_json(res, 'error', 'BAD_INPUT')

def test_post_cve_id_invalid_amount():
    """ cve services rejects amount less than or equal to 0 """
    res = requests.post(
        f'{env.AWG_BASE_URL}{CVE_ID_URL}',
        headers=utils.BASE_HEADERS,
        params={
            'amount': '-1',
            'batch_type': 'sequential',
            'cve_year': f'{utils.CURRENT_YEAR}',
            'short_name': 'mitre'
        }
    )
    assert res.status_code == 400
    response_contains_json(res, 'error', 'INVALID_AMOUNT')

def test_post_cve_id_no_batch_type():
    """ cve services rejects not having a batch type"""
    res = requests.post(
        f'{env.AWG_BASE_URL}{CVE_ID_URL}',
        headers=utils.BASE_HEADERS,
        params={
            'amount': '10',
            'batch_type': '',
            'cve_year': f'{utils.CURRENT_YEAR}',
            'short_name': 'mitre'
        }
    )
    assert res.status_code == 400
    response_contains_json(res, 'error', 'NO_BATCH_TYPE')

def test_post_cve_id_invalid_batch_type():
    """ cve services rejects batch types that aren't 'sequential' or 'nonsequential' """
    res = requests.post(
        f'{env.AWG_BASE_URL}{CVE_ID_URL}',
        headers=utils.BASE_HEADERS,
        params={
            'amount': '10',
            'batch_type': '---',
            'cve_year': f'{utils.CURRENT_YEAR}',
            'short_name': 'mitre'
        }
    )
    assert res.status_code == 400
    response_contains_json(res, 'error', 'INVALID_BATCH_TYPE')

def test_post_cve_id_bad_amount():
    """ api rejects non-numeric amount when requesting IDs """
    res = get_reserve_cve_ids('a', utils.CURRENT_YEAR, 'mitre')
    assert res.status_code == 400
    assert res.reason == 'Bad Request'
    response_contains_json(res, 'error', 'BAD_INPUT')
    assert_contains(res, 'amount')

#### PUT /cve-id/:id ####
def test_put_cve_id_duplicate_state():
    """ api rejects ID updates with duplicate state parameter """
    res = requests.put(
        (f'{env.AWG_BASE_URL}{CVE_ID_URL}/{cve_id}'
         '?state=RESERVED&state=REJECTED'),
        headers=utils.BASE_HEADERS
    )
    assert res.status_code == 400
    assert res.reason == 'Bad Request'
    response_contains_json(res, 'error', 'BAD_INPUT')
    assert_contains(res, 'state')

#### POST /cve-id ####
def test_post_cve_id_reserve_priority():
    """ priority ids can be reserved on behalf of the mitre org """
    res = requests.post(
        f'{env.AWG_BASE_URL}{CVE_ID_URL}',
        headers=utils.BASE_HEADERS,
        params={
            'amount': '1',
            'cve_year': f'{utils.CURRENT_YEAR}',
            'short_name': 'mitre'
        }
    )
    ok_response_contains(res, f'CVE-{utils.CURRENT_YEAR}-')
    assert json.loads(res.content.decode())['cve_ids']
    assert len(json.loads(res.content.decode())['cve_ids']) == 1

    priority_id = json.loads(res.content.decode())['cve_ids'][0]['cve_id']
    assert int(priority_id.split('-')[-1]) < 20000
    # Check that remaining_quota is in response
    for key in json.loads(res.content.decode())['meta'].keys():
        assert key == 'remaining_quota'

# Does this constitute a performance test? The distinction I'm making is that
# performance tests stress the system overall, while these tests try to
# test both "reservation works" and "there's some reasonable amount that we
# can request that doesn't stress the system"

@pytest.mark.parametrize(
    "batch_type, amount",
    [('sequential', 10), ('sequential', 1000),
     ('nonsequential', 1), ('nonsequential', 10)])
def test_post_cve_id_reservation(batch_type, amount):
    """ sequential ids can be reserved on behalf of the mitre org """
    res = get_reserve_cve_ids(amount, utils.CURRENT_YEAR, 'mitre', batch_type)
    ok_response_contains(res, f'CVE-{utils.CURRENT_YEAR}-')
    assert json.loads(res.content.decode())['cve_ids']
    assert len(json.loads(res.content.decode())['cve_ids']) == amount
    # cna and user must exist
    assert 'cna' in res.content.decode()
    assert 'user' in res.content.decode()
    # Check that remaining_quota is in response
    for key in json.loads(res.content.decode())['meta'].keys():
        assert key == 'remaining_quota'

def test_post_cve_id_reserve_sequential_over_quota():
    """ the services api enforces a max quota of 100,000 """
    res = get_reserve_cve_ids(100001, utils.CURRENT_YEAR, 'mitre')
    assert res.status_code == 403
    response_contains_json(res, 'error', 'EXCEEDED_ID_QUOTA')

def test_post_cve_id_reserve_nonsequential_over_limit():
    """ the services api enforces a max non-sequential limit of 10 """
    res = get_reserve_cve_ids(11, utils.CURRENT_YEAR, 'mitre', 'nonsequential')
    assert res.status_code == 403
    response_contains_json(res, 'error', 'OVER_NONSEQUENTIAL_MAX_AMOUNT')

#### GET /cve-id ####
def test_get_cve_id_by_time_reserved():
    """ we can get ids immediately after reserving them using the time they're
    reserved (noting that this may not work against a shared integration
    environment, we check that at least this many have been reserved) """
    n_ids = 10
    time.sleep(1)
    t_before = get_now_timestamp()
    time.sleep(3)
    res_ids = get_reserve_cve_ids(n_ids, utils.CURRENT_YEAR, 'mitre')
    time.sleep(3)
    t_after = get_now_timestamp()

    res_get_ids = requests.get(
        f'{env.AWG_BASE_URL}{CVE_ID_URL}',
        headers=utils.BASE_HEADERS,
        params={
            'time_reserved.lt': t_after,
            'time_reserved.gt': t_before
        }
    )
    ok_response_contains(res_get_ids, f'CVE-{utils.CURRENT_YEAR}-')
    assert len(json.loads(res_get_ids.content.decode())['cve_ids']) == n_ids

def test_get_cve_id_by_time_modified():
    """ we can get ids immediately after reserving them using the time they're
    reserved (noting that this may not work against a shared integration
    environment, we check that at least this many have been reserved) """
    n_ids = 10
    time.sleep(1)
    t_before = get_now_timestamp()
    time.sleep(1)
    res_ids = get_reserve_cve_ids(n_ids, utils.CURRENT_YEAR, 'mitre')
    time.sleep(1)
    t_after = get_now_timestamp()

    res_get_ids = requests.get(
        f'{env.AWG_BASE_URL}{CVE_ID_URL}',
        headers=utils.BASE_HEADERS,
        params={
            'time_modified.lt': t_after,
            'time_modified.gt': t_before
        }
    )
    ok_response_contains(res_get_ids, f'CVE-{utils.CURRENT_YEAR}-')
    assert len(json.loads(res_get_ids.content.decode())['cve_ids']) == n_ids

def test_get_cve_id_empty_parameters():
    """ cannot get id with empty parameters """
    res = requests.get(
        f'{env.AWG_BASE_URL}{CVE_ID_URL}',
        headers=utils.BASE_HEADERS,
        params={
            'page': ' ',
            'state': ' ',
            'cve_id_year': ' ',
            'time_reserved.lt': ' ',
            'time_reserved.gt': ' ',
            'time_modified.lt': ' ',
            'time_modified.gt': ' '
        }
    )
    assert res.status_code == 400
    response_contains(res, 'page')
    response_contains(res, 'state')
    response_contains(res, 'cve_id_year')
    response_contains(res, 'time_reserved.lt')
    response_contains(res, 'time_reserved.gt')
    response_contains(res, 'time_modified.lt')
    response_contains(res, 'time_modified.gt')
    response_contains_json(res, 'error', 'BAD_INPUT')

def test_get_cve_id_page_format_number():
    """ page must be an integer' """
    res = requests.get(
        f'{env.AWG_BASE_URL}{CVE_ID_URL}',
        headers=utils.BASE_HEADERS,
        params={
            'page': 'test',
        }
    )
    assert res.status_code == 400
    response_contains_json(res, 'error', 'BAD_INPUT')

def test_get_cve_id_page_limit():
    """ page must be greater than or equal to 1' """
    res = requests.get(
        f'{env.AWG_BASE_URL}{CVE_ID_URL}',
        headers=utils.BASE_HEADERS,
        params={
            'page': '-1',
        }
    )
    assert res.status_code == 400
    response_contains_json(res, 'error', 'BAD_INPUT')

def test_get_cve_id_state_in_choices():
    """ state parameter can only be 'REJECTED', 'PUBLISHED' or 'RESERVED' """
    res = requests.get(
        f'{env.AWG_BASE_URL}{CVE_ID_URL}',
        headers=utils.BASE_HEADERS,
        params={
            'state': 'TEST',
        }
    )
    assert res.status_code == 400
    response_contains_json(res, 'error', 'BAD_INPUT')

def test_get_cve_id_year_format_with_letters():
    """ cve_id_year format cannot have letters  """
    res = requests.get(
        f'{env.AWG_BASE_URL}{CVE_ID_URL}',
        headers=utils.BASE_HEADERS,
        params={
            'cve_id_year': 'test',
        }
    )
    assert res.status_code == 400
    response_contains_json(res, 'error', 'BAD_INPUT')

def test_get_cve_id_year_format_with_digits():
    """ cve_id_year format must be 4 digits only  """
    res = requests.get(
        f'{env.AWG_BASE_URL}{CVE_ID_URL}',
        headers=utils.BASE_HEADERS,
        params={
            'cve_id_year': '20111',
        }
    )
    assert res.status_code == 400
    response_contains_json(res, 'error', 'BAD_INPUT')

def test_get_cve_id_available_state():
    """ CVE ID filter endpoint does not return any IDs with state 'AVAILABLE' """
    res = requests.get(
        f'{env.AWG_BASE_URL}{CVE_ID_URL}',
        headers=utils.BASE_HEADERS,
        params={
            'page': 1,
            'state': 'PUBLISHED',
            'cve_id_year': 2011
        }
    )
    assert res.status_code == 200
    assert 'AVAILABLE' not in res.content.decode()
    response_contains(res, 'state')

# CVE-ID ENDPOINT UTILITIES
# ==============================================================================
# these are unique to the `{CVE_ID_URL}` endpoint for the AWG system

def get_reserve_cve_ids(
        amount, year, cna_short_name, batch_type='sequential'):
    return requests.post(
        f'{env.AWG_BASE_URL}{CVE_ID_URL}',
        headers=utils.BASE_HEADERS,
        params={
            'amount': f'{amount}',
            'batch_type': batch_type,
            'cve_year': f'{year}',
            'short_name': cna_short_name
        }
    )


text:
# Tests in this file use the secretariat credentials
import requests
import json
import random
import string
from src import env, utils
from src.utils import (assert_contains, ok_response_contains,
                       ok_response_contains_json, response_contains,
                       response_contains_json)

#### GET /users ####
def test_get_all_users():
    """ secretariat users can request a list of all users """
    res = requests.get(
        f'{env.AWG_BASE_URL}/api/users',
        headers=utils.BASE_HEADERS
    )

    test_user={}
    for user in json.loads(res.content.decode())['users']:
        if user['username'] == 'cps@mitre.org':
            test_user = user
            break
    assert test_user['username'] == 'cps@mitre.org'
    assert test_user['name']['first'] == 'Jeremy'
    assert '"secret"' not in res.content.decode() # check that no secrets are included
    assert res.status_code == 200

#### GET /users ####
def test_regular_users_cannot_get_all_users(reg_user_headers):
    """ regular users cannot request a list of all users """
    res = requests.get(
        f'{env.AWG_BASE_URL}/api/users',
        headers=reg_user_headers
    )
    assert res.status_code == 403
    response_contains_json(res, 'error', 'SECRETARIAT_ONLY')

#### GET /users ####
def test_org_admins_cannot_get_all_users(org_admin_headers):
    """ org admins cannot request a list of all users """
    res = requests.get(
        f'{env.AWG_BASE_URL}/api/users',
        headers=org_admin_headers
    )
    assert res.status_code == 403
    response_contains_json(res, 'error', 'SECRETARIAT_ONLY')

def test_get_user_page():
    """ page must be a positive int """
    res = requests.get(
        f'{env.AWG_BASE_URL}/api/users',
        headers=utils.BASE_HEADERS,
        params={
            'page': '1',
        }
    )
    assert res.status_code == 200

def test_get_bad_user_page():
    """ page must be a positive int """

    # test negative
    res = requests.get(
        f'{env.AWG_BASE_URL}/api/users',
        headers=utils.BASE_HEADERS,
        params={
            'page': '-1',
        }
    )
    assert res.status_code == 400
    response_contains_json(res, 'error', 'BAD_INPUT')
    response_contains_json(res, 'details', utils.BAD_PAGE_ERROR_DETAILS)

    # test strings
    res = requests.get(
        f'{env.AWG_BASE_URL}/api/users',
        headers=utils.BASE_HEADERS,
        params={
            'page': 'abc',
        }
    )
    assert res.status_code == 400
    response_contains_json(res, 'error', 'BAD_INPUT')
    response_contains_json(res, 'details', utils.BAD_PAGE_ERROR_DETAILS)

def test_put_user_update_name():
    """ correct error is returned when updating user to same org """

    # grab all users to find one to update
    res = requests.get(
        f'{env.AWG_BASE_URL}/api/users',
        headers=utils.BASE_HEADERS
    )

    assert res.status_code == 200
    all_users = json.loads(res.content.decode())['users']

    assert len(all_users) > 0

    test_user = all_users[0]

    # can only use the org shortname in the URL
    org_res = requests.get(
        f'{env.AWG_BASE_URL}/api/org/{test_user["org_UUID"]}',
        headers=utils.BASE_HEADERS
    )

    assert org_res.status_code == 200
    org = json.loads(org_res.content.decode())

    # random string for test name
    test_name = ''.join(random.choices(string.ascii_letters, k=16))

    res = requests.put(
        f'{env.AWG_BASE_URL}/api/org/{org["short_name"]}/user/{test_user["username"]}',
        headers=utils.BASE_HEADERS,
        params={
            'name.first': test_name
        }
    )

    assert res.status_code == 200

    res = requests.get(
        f'{env.AWG_BASE_URL}/api/org/{org["short_name"]}/user/{test_user["username"]}',
        headers=utils.BASE_HEADERS
    )

    assert json.loads(res.content.decode())['name']['first'] == test_name

    # put the name back to what it was because tests don't reset the data

    res = requests.put(
        f'{env.AWG_BASE_URL}/api/org/{org["short_name"]}/user/{test_user["username"]}',
        headers=utils.BASE_HEADERS,
        params={
            'name.first': test_user['name']['first']
        }
    )

    assert res.status_code == 200

def test_put_user_error_for_same_org():
    """ correct error is returned when updating user to same org """

    # grab all users to find one to update
    res = requests.get(
        f'{env.AWG_BASE_URL}/api/users',
        headers=utils.BASE_HEADERS
    )

    assert res.status_code == 200
    all_users = json.loads(res.content.decode())['users']

    assert len(all_users) > 0

    # try to update user to same org
    test_user = all_users[0]

    # can only use the org shortname in the URL
    org_res = requests.get(
        f'{env.AWG_BASE_URL}/api/org/{test_user["org_UUID"]}',
        headers=utils.BASE_HEADERS
    )

    assert org_res.status_code == 200
    org = json.loads(org_res.content.decode())

    res = requests.put(
        f'{env.AWG_BASE_URL}/api/org/{org["short_name"]}/user/{test_user["username"]}',
        headers=utils.BASE_HEADERS,
        params={
            'org_short_name': org['short_name']
        }
    )

    assert res.status_code == 403
    err = json.loads(res.content.decode())

    assert err['error'] == 'USER_ALREADY_IN_ORG'
    assert err['message'] == f'The user could not be updated because the user \'{test_user["username"]}\' already belongs to the \'{org["short_name"]}\' organization.'