/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.daffodil.general

import org.junit.Test
import org.junit.Assert.assertEquals
import org.junit.Assert.assertFalse
import org.junit.Assert.assertTrue

import org.apache.daffodil.Implicits.intercept
import org.apache.daffodil.dpath.NodeInfo
import org.apache.daffodil.infoset.DISimple
import org.apache.daffodil.infoset.ScalaXMLInfosetInputter
import org.apache.daffodil.infoset.ScalaXMLInfosetOutputter
import org.apache.daffodil.io.InputSourceDataInputStream
import org.apache.daffodil.util.SchemaUtils
import org.apache.daffodil.util.TestUtils

/**
 * Test InfosetOutputter that extends the ScalaXMLInfosetOutputter and redacts
 * words specified in the dfdlx:runtimeProperties extension
 */
class RedactingScalaXMLInfosetOutputter
  extends ScalaXMLInfosetOutputter {

  override def startSimple(diSimple: DISimple): Unit = {
    super.startSimple(diSimple)

    val runtimeProperties = diSimple.erd.runtimeProperties

    val redactions = Option(runtimeProperties.get("redact")).map { value => value.split(",") }
    if (redactions.isDefined) {
      val thisElem = stack.top.last.asInstanceOf[scala.xml.Elem]
      var text = thisElem.child(0).asInstanceOf[scala.xml.Text].data
      val redactWord = runtimeProperties.getOrDefault("redactWord", "")
      redactions.get.foreach { redaction =>
        text = text.replace(redaction, redactWord)
      }
      val newChild = Seq(new scala.xml.Text(text))
      val newElem = thisElem.copy(child = newChild)
      stack.top(stack.top.size - 1) = newElem
    }
  }
}

/**
 * Test InfosetInputter that extends the ScalaXMLInfosetInputter and redacts
 * words specified in the dfdlx:runtimeProperties extension
 */
class RedactingScalaXMLInfosetInputter(rootNode: scala.xml.Node)
  extends ScalaXMLInfosetInputter(rootNode) {

  override def getSimpleText(primType: NodeInfo.Kind, runtimeProperties: java.util.Map[String, String]): String = {
    var text = super.getSimpleText(primType, runtimeProperties)

    val redactions = Option(runtimeProperties.get("redact")).map { value => value.split(",") }
    if (redactions.isDefined) {
      val redactWord = runtimeProperties.getOrDefault("redactWord", "")
      redactions.get.foreach { redaction =>
        text = text.replace(redaction, redactWord)
      }
    }
    text
  }
}

class TestRuntimeProperties {

  val testSchema1 = SchemaUtils.dfdlTestSchema(
    <xs:include schemaLocation="org/apache/daffodil/xsd/DFDLGeneralFormat.dfdl.xsd"/>,
    <dfdl:format ref="tns:GeneralFormat" lengthKind="delimited"/>,
    <xs:element name="root">
      <xs:complexType>
        <xs:sequence dfdl:separator=",">
          <xs:element name="filtered" type="xs:string" dfdlx:runtimeProperties="redact=foo,baz redactWord=REDACTED" />
          <xs:element name="unFiltered" type="xs:string" />
        </xs:sequence>
      </xs:complexType>
    </xs:element>
  )

  @Test def testRuntimeProperties_01(): Unit = {

    val data = "foo bar baz,foo bar baz"
    val expected =
      <root xmlns="http://example.com">
        <filtered>REDACTED bar REDACTED</filtered>
        <unFiltered>foo bar baz</unFiltered>
      </root>

    val input = InputSourceDataInputStream(data.getBytes("UTF-8"))
    val outputter = new RedactingScalaXMLInfosetOutputter()

    val dp = TestUtils.compileSchema(testSchema1)
    val pr = dp.parse(input, outputter)

    assertFalse(pr.isError)
    val actual = outputter.getResult()
    TestUtils.assertEqualsXMLElements(expected, actual)
  }

  @Test def testRuntimeProperties_02(): Unit = {

    val infoset =
      <root xmlns="http://example.com">
        <filtered>foo bar baz</filtered>
        <unFiltered>foo bar baz</unFiltered>
      </root>
    val expected = "REDACTED bar REDACTED,foo bar baz"

    val inputter = new RedactingScalaXMLInfosetInputter(infoset)
    val output = new java.io.ByteArrayOutputStream()

    val dp = TestUtils.compileSchema(testSchema1)
    val ur = dp.unparse(inputter, output)

    assertFalse(ur.isError)
    
    val actual = output.toString("UTF-8")
    assertEquals(expected, actual)
  }

  val testSchemaBad1 = SchemaUtils.dfdlTestSchema(
    <xs:include schemaLocation="org/apache/daffodil/xsd/DFDLGeneralFormat.dfdl.xsd"/>,
    <dfdl:format ref="tns:GeneralFormat" lengthKind="delimited"/>,
    <xs:element name="root" dfdlx:runtimeProperties="redact=foo,baz">
      <xs:complexType>
        <xs:sequence dfdl:separator=",">
          <xs:element name="filtered" type="xs:string" />
          <xs:element name="unFiltered" type="xs:string" />
        </xs:sequence>
      </xs:complexType>
    </xs:element>
  )

  @Test def testRuntimeProperties_03(): Unit = {
    val ex = intercept[Exception] {
      TestUtils.compileSchema(testSchemaBad1)
    }
    val msg = ex.getMessage()

    assertTrue(msg.contains("Schema Definition Error"))
    assertTrue(msg.contains("dfdlx:runtimeProperties"))
    assertTrue(msg.contains("simple type"))
  }

  val testSchemaBad2 = SchemaUtils.dfdlTestSchema(
    <xs:include schemaLocation="org/apache/daffodil/xsd/DFDLGeneralFormat.dfdl.xsd"/>,
    <dfdl:format ref="tns:GeneralFormat" lengthKind="delimited"/>,
    <xs:element name="root">
      <xs:complexType>
        <xs:sequence dfdl:separator=",">
          <xs:element name="filtered" type="xs:string" dfdlx:runtimeProperties="redact=" />
          <xs:element name="unFiltered" type="xs:string" />
        </xs:sequence>
      </xs:complexType>
    </xs:element>
  )

  @Test def testRuntimeProperties_04(): Unit = {
    val ex = intercept[Exception] {
      TestUtils.compileSchema(testSchemaBad2)
    }
    val msg = ex.getMessage()

    assertTrue(msg.contains("Schema Definition Error"))
    assertTrue(msg.contains("dfdlx:runtimeProperties"))
    assertTrue(msg.contains("key=value"))
  }

  val testSchema2 = SchemaUtils.dfdlTestSchema(
    <xs:include schemaLocation="org/apache/daffodil/xsd/DFDLGeneralFormat.dfdl.xsd"/>,
    <dfdl:format ref="tns:GeneralFormat" lengthKind="delimited"/>,
    <xs:element name="root">
      <xs:complexType>
        <xs:sequence dfdl:separator=",">
          <xs:element name="filtered" type="xs:string">
            <xs:annotation>
              <xs:appinfo source="http://www.ogf.org/dfdl/">
                <dfdl:element>
                  <dfdl:property name="dfdlx:runtimeProperties">
                    redact=foo,baz
                    redactWord=REDACTED
                  </dfdl:property>
                </dfdl:element>
              </xs:appinfo>
            </xs:annotation>
          </xs:element>
          <xs:element name="unFiltered" type="xs:string" />
        </xs:sequence>
      </xs:complexType>
    </xs:element>
  )

  @Test def testRuntimeProperties_05(): Unit = {

    val data = "foo bar baz,foo bar baz"
    val expected =
      <root xmlns="http://example.com">
        <filtered>REDACTED bar REDACTED</filtered>
        <unFiltered>foo bar baz</unFiltered>
      </root>

    val input = InputSourceDataInputStream(data.getBytes("UTF-8"))
    val outputter = new RedactingScalaXMLInfosetOutputter()

    val dp = TestUtils.compileSchema(testSchema2)
    val pr = dp.parse(input, outputter)

    assertFalse(pr.isError)
    val actual = outputter.getResult()
    TestUtils.assertEqualsXMLElements(expected, actual)
  }

}
